test(logmq): fail fast when batcher.Shutdown deadlocks

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()

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)

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 {

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)