feat(logmq): parallelize the post-persist pipeline, one goroutine per entry

After the batched insert, each entry's pipeline now runs on its own goroutine
— dispatch-and-move-on — and an attempt's events are sent concurrently under
a 5s emit timeout. A slow eval or sink send no longer blocks persisting the
next batch, and one destination's slow sink no longer serializes everyone
else's sends.

Entries process in no particular order, including within a destination
(discussed on #980): the consecutive-failure counter is a set of attempt IDs,
so counting is idempotent and commutative, and per-process ordering was
cosmetic anyway with multiple logmq replicas. No pools, queues, or sizing
knobs: in-flight work is naturally bounded by arrival rate × the emit
timeout, and Shutdown drains the in-flight goroutines (bounded by the same
timeout).

Characterization tests updated for the ordering relaxation: assertions pin
topic multisets and disable calls, never arrival order or attempt identity;
tests that need a deterministic sequence (success-reset keystone, suppression
window) pace messages one at a time. New tests pin the decoupling (slow
send/eval doesn't block persistence), same-destination independence, and
shutdown draining.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: alexluong

internal/logmq/batchprocessor.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/hookdeck/outpost/internal/alert"
@@ -13,11 +14,19 @@ import (
 	"github.com/hookdeck/outpost/internal/opevents"
 	"github.com/mikestefanello/batcher"
 	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
 )
 
 // ErrInvalidLogEntry is returned when a LogEntry is missing required fields.
 var ErrInvalidLogEntry = errors.New("invalid log entry: both event and attempt are required")
 
+// emitTimeout caps a single sink send. It is the system's definition of the
+// worst acceptable send latency, and it bounds each entry goroutine's
+// lifetime: in-flight work steady-states at arrival rate × this timeout even
+// when every send is stuck, and the shutdown drain is bounded by it. Anything
+// slower fails into the nack/redelivery path.
+const emitTimeout = 5 * time.Second
+
 // LogStore defines the interface for persisting log entries.
 // This is a consumer-defined interface containing only what logmq needs.
 type LogStore interface {
@@ -82,14 +91,25 @@ type BatchProcessorConfig struct {
 }
 
 // BatchProcessor batches log entries and writes them to the log store, then
-// runs the alert pipeline per entry — evaluate, act on the verdict (disable,
-// operator events), dedup replays — serially in the batch loop.
+// runs the alert pipeline per entry on its own goroutine — dispatch-and-move-on,
+// so a slow eval or sink send never blocks persistence of the next batch.
+//
+// Entries process in no particular order, including within a destination: the
+// consecutive-failure count tolerates approximate order (its store is a set of
+// attempt IDs, so counting is idempotent and commutative; only a success/
+// failure race around the reset can skew it), and per-process ordering was
+// cosmetic anyway with multiple logmq replicas interleaving a destination's
+// attempts. See the discussion on the parallelism RFC.
 type BatchProcessor struct {
 	ctx      context.Context
 	logger   *logging.Logger
 	logStore LogStore
 	alerts   AlertPipeline
 	batcher  *batcher.Batcher[*mqs.Message]
+	// inflight tracks entry goroutines so Shutdown can drain them. Each is
+	// bounded by emitTimeout, so the wait is bounded too.
+	inflight     sync.WaitGroup
+	shutdownOnce sync.Once
 }
 
 // NewBatchProcessor creates a new batch processor for log entries.
@@ -130,11 +150,16 @@ func (bp *BatchProcessor) Add(ctx context.Context, msg *mqs.Message) error {
 	return nil
 }
 
-// Shutdown gracefully shuts down the batch processor. The batcher flushes
-// pending batches, so every buffered message reaches a terminal state before
-// Shutdown returns.
+// Shutdown gracefully shuts down the batch processor: the batcher first
+// (flushes pending batches, which may still dispatch entry goroutines), then
+// the in-flight entries drain. Every dispatched message reaches a terminal
+// state before Shutdown returns, and the drain is bounded by emitTimeout.
+// Idempotent.
 func (bp *BatchProcessor) Shutdown() {
-	bp.batcher.Shutdown()
+	bp.shutdownOnce.Do(func() {
+		bp.batcher.Shutdown()
+		bp.inflight.Wait()
+	})
 }
 
 // processBatch processes a batch of messages.
@@ -227,7 +252,11 @@ func (bp *BatchProcessor) processBatch(_ string, msgs []*mqs.Message) {
 		zap.Int("count", len(validMsgs)),
 		zap.Int64("insert_duration_ms", time.Since(insertStart).Milliseconds()))
 
-	// Run the alert pipeline per persisted entry.
+	// Spawn one goroutine per persisted entry and return — the batch loop
+	// never waits on eval or delivery. In-flight goroutines are bounded by
+	// arrival rate × emitTimeout (each lives at most about one send latency),
+	// and every fetched message reaches ack/nack well inside the broker's
+	// visibility window.
 	for i, entry := range entries {
 		if bp.alerts.Evaluator == nil {
 			validMsgs[i].Ack()
@@ -243,7 +272,10 @@ func (bp *BatchProcessor) processBatch(_ string, msgs []*mqs.Message) {
 			continue
 		}
 
-		bp.processEntry(bp.ctx, entry, validMsgs[i])
+		msg := validMsgs[i]
+		bp.inflight.Go(func() {
+			bp.processEntry(bp.ctx, entry, msg)
+		})
 	}
 }
 
@@ -302,20 +334,30 @@ func (bp *BatchProcessor) processEntry(ctx context.Context, entry *models.LogEnt
 		return
 	}
 
-	// Deliver the attempt's events. Any failure nacks with nothing marked, so
+	// Deliver the attempt's events concurrently; arrival order within an
+	// attempt is not guaranteed. Any failure nacks with nothing marked, so
 	// redelivery re-runs the attempt in full — events already sent may go out
 	// again (at-least-once).
+	g, gctx := errgroup.WithContext(ctx)
 	for _, de := range events {
-		if err := bp.send(ctx, de, entry); err != nil {
-			bp.logger.Ctx(ctx).Error("opevent delivery failed",
-				zap.Error(err),
-				zap.String("topic", de.event.Topic),
-				zap.String("attempt_id", entry.Attempt.ID),
-				zap.String("event_id", entry.Event.ID),
-				zap.String("destination_id", entry.Destination.ID))
-			msg.Nack()
-			return
-		}
+		g.Go(func() error {
+			sendCtx, cancel := context.WithTimeout(gctx, emitTimeout)
+			defer cancel()
+			if err := bp.send(sendCtx, de, entry); err != nil {
+				bp.logger.Ctx(ctx).Error("opevent delivery failed",
+					zap.Error(err),
+					zap.String("topic", de.event.Topic),
+					zap.String("attempt_id", entry.Attempt.ID),
+					zap.String("event_id", entry.Event.ID),
+					zap.String("destination_id", entry.Destination.ID))
+				return err
+			}
+			return nil
+		})
+	}
+	if g.Wait() != nil {
+		msg.Nack()
+		return
 	}
 
 	if err := bp.alerts.ProcessedIdemp.MarkProcessed(ctx, key); err != nil {
@@ -338,10 +380,11 @@ type deliveryEvent struct {
 }
 
 // plan acts on an evaluation and builds the operator events owed for this
-// attempt — disabled, consecutive_failure, exhausted_retries, sent in slice
-// order. The disable (a DB write) happens here: it's an action, not a
-// notification, and it must precede event construction so the payloads carry
-// the destination's latest state (disabled).
+// attempt — disabled, consecutive_failure, exhausted_retries. They are sent
+// concurrently, so slice order carries no meaning. The disable (a DB write)
+// happens here: it's an action, not a notification, and it must precede event
+// construction so the payloads carry the destination's latest state
+// (disabled).
 func (bp *BatchProcessor) plan(ctx context.Context, eval alert.Evaluation, entry *models.LogEntry) ([]deliveryEvent, error) {
 	if eval.ConsecutiveFailure == nil && !eval.RetriesExhausted {
 		return nil, nil

internal/logmq/batchprocessor_test.go

@@ -48,7 +48,7 @@ func (m *mockLogStore) getInserted() (events []*models.Event, attempts []*models
 }
 
 // mockQueueMessage implements mqs.QueueMessage for testing. Terminal state is
-// atomic: acks/nacks land off the test goroutine.
+// atomic: acks/nacks land on per-entry goroutines.
 type mockQueueMessage struct {
 	acked  atomic.Bool
 	nacked atomic.Bool

internal/logmq/characterization_decoupling_test.go

@@ -0,0 +1,122 @@
+package logmq_test
+
+// delivery. Each entry delivers on its own goroutine, so a slow sink stalls only
+// the messages that owe events, never the batch loop.
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/hookdeck/outpost/internal/models"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// A batch-1 attempt whose alert delivery hangs must not delay batch 2's
+// persistence or ack. (If sends ran serially in the batch loop, batch 2 would
+// wait on batch 1's sink call.)
+func TestCharacterization_SlowDeliveryDoesNotBlockPersistence(t *testing.T) {
+	t.Parallel()
+	h := newHarness(t, harnessConfig{
+		batcher: batcherConfig{itemCount: 1},
+		// max=2, thresholds[50,100] → a single failure (count 1 = 50%) alerts.
+		alert: alertConfig{autoDisableCount: 2, thresholds: []int{50, 100}},
+		doubles: doublesConfig{
+			sinkBlockOn: map[string]bool{"att_slow": true},
+		},
+	})
+
+	tenant := "tenant_d1"
+
+	// Batch 1: alerting failure whose send blocks.
+	cmSlow, msgSlow := newCountingMessage(makeEntry("dest_d1_slow", tenant, "att_slow", models.AttemptStatusFailed))
+	h.add(msgSlow)
+
+	// Wait until the delivery is actually blocked inside the sink.
+	require.Eventually(t, func() bool { return h.sink.inflightSends() >= 1 },
+		5*time.Second, 5*time.Millisecond, "the slow delivery should reach the sink")
+
+	// Batch 2: a plain success — persists and acks while batch 1's send hangs.
+	cmOK, msgOK := newCountingMessage(makeEntry("dest_d1_ok", tenant, "att_ok", models.AttemptStatusSuccess))
+	h.add(msgOK)
+	h.waitTerminal([]*countingMessage{cmOK})
+	cmOK.requireAcked(t)
+	require.Len(t, h.listAttempt("dest_d1_ok"), 1, "batch 2 persisted while batch 1's delivery hangs")
+
+	// The slow message is still in flight: persisted but no terminal state.
+	require.Len(t, h.listAttempt("dest_d1_slow"), 1, "the slow attempt itself persisted immediately")
+	assert.EqualValues(t, 0, cmSlow.acks()+cmSlow.nacks(), "the slow message stays un-acked until its delivery completes")
+
+	// Release the sink: the blocked delivery completes and acks.
+	h.sink.release()
+	h.waitTerminal([]*countingMessage{cmSlow})
+	cmSlow.requireAcked(t)
+	assert.Equal(t, []string{topicCF}, topics(h.sink.forDest("dest_d1_slow")))
+}
+
+// A hot destination's alerting attempts deliver in parallel — one slow send
+// doesn't serialize the destination's other sends (the old per-dest serial
+// loop did exactly that).
+func TestCharacterization_HotDestinationDeliversInParallel(t *testing.T) {
+	t.Parallel()
+	h := newHarness(t, harnessConfig{
+		batcher: batcherConfig{itemCount: 3},
+		// max=100 with thresholds 1/2/3% → counts 1, 2 and 3 each cross one
+		// threshold, so all three attempts alert.
+		alert: alertConfig{autoDisableCount: 100, thresholds: []int{1, 2, 3}},
+		doubles: doublesConfig{
+			sinkBlockOn: map[string]bool{topicCF: true}, // every cf send blocks
+		},
+	})
+
+	destA, tenant := "dest_d2", "tenant_d2"
+	msgs := make([]*countingMessage, 0, 3)
+	for i := 1; i <= 3; i++ {
+		cm, msg := newCountingMessage(makeEntry(destA, tenant, fmt.Sprintf("att_%d", i), models.AttemptStatusFailed))
+		msgs = append(msgs, cm)
+		h.add(msg)
+	}
+
+	// All three sends block in the sink AT THE SAME TIME — same destination,
+	// different goroutines.
+	require.Eventually(t, func() bool { return h.sink.inflightSends() >= 3 },
+		5*time.Second, 5*time.Millisecond, "the destination's deliveries should run concurrently")
+
+	h.sink.release()
+	h.waitTerminal(msgs)
+	for _, m := range msgs {
+		m.requireAcked(t)
+	}
+	assert.GreaterOrEqual(t, h.sink.maxInflightSends(), int32(3))
+	assert.ElementsMatch(t, []string{"att_1", "att_2", "att_3"}, attemptIDs(h.sink.forDest(destA)))
+}
+
+// Shutdown drains in-flight deliveries: every dispatched entry reaches its
+// terminal state before Shutdown returns.
+func TestCharacterization_ShutdownDrainsDeliveries(t *testing.T) {
+	t.Parallel()
+	h := newHarness(t, harnessConfig{
+		batcher: batcherConfig{itemCount: 1},
+		alert:   alertConfig{autoDisableCount: 2, thresholds: []int{50, 100}},
+		doubles: doublesConfig{
+			sinkBlockOn: map[string]bool{"att_drain": true},
+		},
+	})
+
+	cm, msg := newCountingMessage(makeEntry("dest_d3", "tenant_d3", "att_drain", models.AttemptStatusFailed))
+	h.add(msg)
+	require.Eventually(t, func() bool { return h.sink.inflightSends() >= 1 },
+		5*time.Second, 5*time.Millisecond)
+
+	// Release while Shutdown is waiting on the drain.
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		h.sink.release()
+	}()
+	h.bp.Shutdown()
+
+	// Shutdown returned → the delivery completed and acked.
+	cm.requireAcked(t)
+	assert.Equal(t, []string{topicCF}, topics(h.sink.forDest("dest_d3")))
+}

internal/logmq/characterization_harness_test.go

@@ -15,14 +15,19 @@ package logmq_test
 //   - per-message ack/nack counters (exactly-once terminal state)
 //   - logStore.ListAttempt / ListEvent
 //
-// Per-destination order only — never assert global cross-destination order.
+// Entries process concurrently and in no particular order (goroutine per
+// entry), so never assert arrival order — across destinations OR within one.
+// Tests whose semantics need a deterministic sequence pace it themselves:
+// add one message, waitTerminal, add the next.
 //
 // Files in this suite (concern → file):
 //   - characterization_harness_test.go         shared setup, doubles, helpers
-//   - characterization_ordering_test.go        eval ordering & counting
+//   - characterization_ordering_test.go        failure counting & thresholds
 //   - characterization_idempotency_test.go     replay / idempotency
 //   - characterization_acknowledgement_test.go ack/nack exactly-once
 //   - characterization_validation_test.go      intake (parse / dedup / persist)
+//   - characterization_decoupling_test.go      persistence decoupled from delivery
+//   - characterization_postprocess_test.go     post-persist eval concurrency
 
 import (
 	"context"