From c1519cf438c7ef73285d83cb6cc56d23ee20c5a0 Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Fri, 14 Aug 2026 17:14:23 +0700 Subject: [PATCH 1/2] fix(logmq): bound BatchProcessor.Shutdown batcher.Shutdown can deadlock: it stops the ticker, then calls processQueue, whose defer restarts it. If a tick lands between the processingMutex acquisition and the send on the shutdown channel, the ticker goroutine blocks on that mutex forever and the send never finds a receiver. Both goroutines are stuck for the process lifetime. Shutdown now runs the drain on its own goroutine and gives up after 60s, above the legitimate worst case of a 30s final insert plus one emitTimeout drain. Co-Authored-By: Claude Opus 5 (1M context) --- internal/logmq/batchprocessor.go | 30 ++++++++++++++++++++++++++++-- 1 file changed, 28 insertions(+), 2 deletions(-) diff --git a/internal/logmq/batchprocessor.go b/internal/logmq/batchprocessor.go index 281c98c3..957f5502 100644 --- a/internal/logmq/batchprocessor.go +++ b/internal/logmq/batchprocessor.go @@ -28,6 +28,16 @@ var ErrInvalidLogEntry = errors.New("invalid log entry: both event and attempt a // slower fails into the nack/redelivery path. const emitTimeout = 5 * time.Second +// shutdownTimeout caps Shutdown so a stuck drain cannot hold the process open +// indefinitely. It sits above the legitimate worst case — the final flush's +// 30s insert plus one emitTimeout drain — so it only ever fires on a hang. +// +// The known hang is in batcher.Shutdown: it stops the ticker, then calls +// processQueue, whose defer restarts it. If a tick lands between the +// processingMutex acquisition and the send on its shutdown channel, the ticker +// goroutine blocks on that mutex forever and the send never finds a receiver. +const shutdownTimeout = 60 * time.Second + // LogStore defines the interface for persisting log entries. // This is a consumer-defined interface containing only what logmq needs. type LogStore interface { @@ -181,10 +191,26 @@ func (bp *BatchProcessor) Add(ctx context.Context, msg *mqs.Message) error { // the in-flight entries drain. Every dispatched message reaches a terminal // state before Shutdown returns, and the drain is bounded by emitTimeout. // Idempotent. +// +// Shutdown gives up after shutdownTimeout and returns, abandoning the drain +// goroutine. Messages still in flight are never acked, so the broker +// redelivers them. Losing the process to a stuck drain would strand them the +// same way, with the process hanging until it is killed. func (bp *BatchProcessor) Shutdown() { bp.shutdownOnce.Do(func() { - bp.batcher.Shutdown() - bp.inflight.Wait() + done := make(chan struct{}) + go func() { + defer close(done) + bp.batcher.Shutdown() + bp.inflight.Wait() + }() + + select { + case <-done: + case <-time.After(shutdownTimeout): + bp.logger.Ctx(bp.ctx).Error("logmq batch processor shutdown timed out", + zap.Duration("timeout", shutdownTimeout)) + } }) } From 90339cc19e98cf11c7b1328b7863763999fc8e7c Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Fri, 14 Aug 2026 17:18:53 +0700 Subject: [PATCH 2/2] test(logmq): bound Shutdown in tests, revert the production change Keep the production drain unbounded. The hang is rare enough that changing shutdown behavior to defend against it is not worth it; the only real cost is a 10m CI timeout, so bound it where that cost lands. shutdownBounded fails the test after 30s instead. The package runs in a few seconds, so nothing legitimate is near that. Co-Authored-By: Claude Opus 5 (1M context) --- internal/logmq/batchprocessor.go | 30 ++----------------- internal/logmq/batchprocessor_test.go | 18 +++++------ .../logmq/characterization_decoupling_test.go | 2 +- .../logmq/characterization_harness_test.go | 30 ++++++++++++++++++- .../characterization_postprocess_test.go | 2 +- 5 files changed, 42 insertions(+), 40 deletions(-) diff --git a/internal/logmq/batchprocessor.go b/internal/logmq/batchprocessor.go index 957f5502..281c98c3 100644 --- a/internal/logmq/batchprocessor.go +++ b/internal/logmq/batchprocessor.go @@ -28,16 +28,6 @@ var ErrInvalidLogEntry = errors.New("invalid log entry: both event and attempt a // slower fails into the nack/redelivery path. const emitTimeout = 5 * time.Second -// shutdownTimeout caps Shutdown so a stuck drain cannot hold the process open -// indefinitely. It sits above the legitimate worst case — the final flush's -// 30s insert plus one emitTimeout drain — so it only ever fires on a hang. -// -// The known hang is in batcher.Shutdown: it stops the ticker, then calls -// processQueue, whose defer restarts it. If a tick lands between the -// processingMutex acquisition and the send on its shutdown channel, the ticker -// goroutine blocks on that mutex forever and the send never finds a receiver. -const shutdownTimeout = 60 * time.Second - // LogStore defines the interface for persisting log entries. // This is a consumer-defined interface containing only what logmq needs. type LogStore interface { @@ -191,26 +181,10 @@ func (bp *BatchProcessor) Add(ctx context.Context, msg *mqs.Message) error { // the in-flight entries drain. Every dispatched message reaches a terminal // state before Shutdown returns, and the drain is bounded by emitTimeout. // Idempotent. -// -// Shutdown gives up after shutdownTimeout and returns, abandoning the drain -// goroutine. Messages still in flight are never acked, so the broker -// redelivers them. Losing the process to a stuck drain would strand them the -// same way, with the process hanging until it is killed. func (bp *BatchProcessor) Shutdown() { bp.shutdownOnce.Do(func() { - done := make(chan struct{}) - go func() { - defer close(done) - bp.batcher.Shutdown() - bp.inflight.Wait() - }() - - select { - case <-done: - case <-time.After(shutdownTimeout): - bp.logger.Ctx(bp.ctx).Error("logmq batch processor shutdown timed out", - zap.Duration("timeout", shutdownTimeout)) - } + bp.batcher.Shutdown() + bp.inflight.Wait() }) } diff --git a/internal/logmq/batchprocessor_test.go b/internal/logmq/batchprocessor_test.go index cd9b4f2a..8e17c749 100644 --- a/internal/logmq/batchprocessor_test.go +++ b/internal/logmq/batchprocessor_test.go @@ -88,7 +88,7 @@ func TestBatchProcessor_ValidEntry(t *testing.T) { DelayThreshold: 1 * time.Second, }) require.NoError(t, err) - defer bp.Shutdown() + defer shutdownBounded(t, bp) event := testutil.EventFactory.Any() attempt := testutil.AttemptFactory.Any() @@ -122,7 +122,7 @@ func TestBatchProcessor_InvalidEntry_MissingEvent(t *testing.T) { DelayThreshold: 1 * time.Second, }) require.NoError(t, err) - defer bp.Shutdown() + defer shutdownBounded(t, bp) attempt := testutil.AttemptFactory.Any() entry := models.LogEntry{ @@ -155,7 +155,7 @@ func TestBatchProcessor_InvalidEntry_MissingAttempt(t *testing.T) { DelayThreshold: 1 * time.Second, }) require.NoError(t, err) - defer bp.Shutdown() + defer shutdownBounded(t, bp) event := testutil.EventFactory.Any() entry := models.LogEntry{ @@ -188,7 +188,7 @@ func TestBatchProcessor_InvalidEntry_DoesNotBlockBatch(t *testing.T) { DelayThreshold: 1 * time.Second, }) require.NoError(t, err) - defer bp.Shutdown() + defer shutdownBounded(t, bp) // Create valid entry 1 event1 := testutil.EventFactory.Any() @@ -244,7 +244,7 @@ func TestBatchProcessor_DuplicateMessages(t *testing.T) { DelayThreshold: 1 * time.Second, }) require.NoError(t, err) - defer bp.Shutdown() + defer shutdownBounded(t, bp) // Two byte-identical copies of the same entry (redelivery / re-publish) event := testutil.EventFactory.Any() @@ -297,7 +297,7 @@ func TestBatchProcessor_MalformedJSON(t *testing.T) { DelayThreshold: 1 * time.Second, }) require.NoError(t, err) - defer bp.Shutdown() + defer shutdownBounded(t, bp) mock, msg := newMockMessageFromBytes([]byte("not valid json")) err = bp.Add(ctx, msg) @@ -363,7 +363,7 @@ func TestBatchProcessor_AlertEvaluator_WithDestination(t *testing.T) { DelayThreshold: 1 * time.Second, }) require.NoError(t, err) - defer bp.Shutdown() + defer shutdownBounded(t, bp) event := testutil.EventFactory.Any() attempt := testutil.AttemptFactory.Any() @@ -398,7 +398,7 @@ func TestBatchProcessor_AlertEvaluator_NilDestination(t *testing.T) { DelayThreshold: 1 * time.Second, }) require.NoError(t, err) - defer bp.Shutdown() + defer shutdownBounded(t, bp) event := testutil.EventFactory.Any() attempt := testutil.AttemptFactory.Any() @@ -429,7 +429,7 @@ func TestBatchProcessor_AlertEvaluator_Error(t *testing.T) { DelayThreshold: 1 * time.Second, }) require.NoError(t, err) - defer bp.Shutdown() + defer shutdownBounded(t, bp) event := testutil.EventFactory.Any() attempt := testutil.AttemptFactory.Any() diff --git a/internal/logmq/characterization_decoupling_test.go b/internal/logmq/characterization_decoupling_test.go index 2d752066..f0a47b0b 100644 --- a/internal/logmq/characterization_decoupling_test.go +++ b/internal/logmq/characterization_decoupling_test.go @@ -115,7 +115,7 @@ func TestCharacterization_ShutdownDrainsDeliveries(t *testing.T) { time.Sleep(50 * time.Millisecond) h.sink.release() }() - h.bp.Shutdown() + shutdownBounded(t, h.bp) // Shutdown returned → the delivery completed and acked. cm.requireAcked(t) diff --git a/internal/logmq/characterization_harness_test.go b/internal/logmq/characterization_harness_test.go index 49d9ae07..2939858d 100644 --- a/internal/logmq/characterization_harness_test.go +++ b/internal/logmq/characterization_harness_test.go @@ -182,6 +182,34 @@ func (e *blockingEvaluator) release() { func (e *blockingEvaluator) blockedEvals() int32 { return e.blocked.Load() } func (e *blockingEvaluator) enteredEvals() int32 { return e.entered.Load() } +// shutdownGrace bounds bp.Shutdown in tests. Nothing here legitimately takes +// this long — the whole package runs in a few seconds. +const shutdownGrace = 30 * time.Second + +// shutdownBounded calls bp.Shutdown and fails the test if it does not return +// within shutdownGrace, instead of riding the 10m package timeout. +// +// batcher.Shutdown (mikestefanello/batcher@v0.1.0) can deadlock: it stops the +// ticker, then calls processQueue, whose defer restarts it. A tick landing +// between the processingMutex acquisition and the send on its shutdown channel +// strands the ticker goroutine on that mutex, leaving the send with no +// receiver. Rare, but it takes the whole package down with it when it happens. +func shutdownBounded(t *testing.T, bp *logmq.BatchProcessor) { + t.Helper() + + done := make(chan struct{}) + go func() { + defer close(done) + bp.Shutdown() + }() + + select { + case <-done: + case <-time.After(shutdownGrace): + t.Errorf("bp.Shutdown did not return within %s", shutdownGrace) + } +} + type disableRecord struct { tenantID string destinationID string @@ -411,7 +439,7 @@ func newHarness(t *testing.T, cfg harnessConfig) *harness { EmitTimeout: cfg.batcher.emitTimeout, }) require.NoError(t, err) - t.Cleanup(bp.Shutdown) + t.Cleanup(func() { shutdownBounded(t, bp) }) // LIFO: releases run BEFORE bp.Shutdown, so a test that never released its // blocked sends/evals can't deadlock the drain. if sink.blockCh != nil { diff --git a/internal/logmq/characterization_postprocess_test.go b/internal/logmq/characterization_postprocess_test.go index cfca7b62..598d8268 100644 --- a/internal/logmq/characterization_postprocess_test.go +++ b/internal/logmq/characterization_postprocess_test.go @@ -111,7 +111,7 @@ func TestCharacterization_ShutdownDrainsPostprocess(t *testing.T) { time.Sleep(50 * time.Millisecond) h.eval.release() }() - h.bp.Shutdown() + shutdownBounded(t, h.bp) // Shutdown returned → the eval ran, the alert delivered and the msg acked. cm.requireAcked(t)