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)