From 20314e1ccd41c8642f326eb093f0908b3ddea28e Mon Sep 17 00:00:00 2001
From: Alex Luong <alex@alexluong.com>
Date: Sun, 25 Oct 2026 00:18:14 +0700
Subject: [PATCH 1/2] fix: take a consumer handler slot before receiving

---
 internal/consumer/consumer.go      | 17 ++++++++-----
 internal/consumer/consumer_test.go | 41 ++++++++++++++++++++++++++++++
 2 files changed, 51 insertions(+), 7 deletions(-)

diff --git a/internal/consumer/consumer.go b/internal/consumer/consumer.go
index a3ce0ed42..233ad099a 100644
--- a/internal/consumer/consumer.go
+++ b/internal/consumer/consumer.go
@@ -205,7 +205,9 @@ recvLoop:
 	return subscriptionErr
 }
 
-// runWithSemaphore limits concurrency via a channel-based semaphore.
+// runWithSemaphore limits concurrency via a channel-based semaphore. The slot
+// is taken before receiving, so a message is only received when a handler can
+// run it now.
 func (c *consumerImpl) runWithSemaphore(ctx context.Context) error {
 	tracer := otel.GetTracerProvider().Tracer("github.com/hookdeck/outpost/internal/consumer")
 
@@ -215,18 +217,19 @@ func (c *consumerImpl) runWithSemaphore(ctx context.Context) error {
 	sem := make(chan struct{}, c.concurrency)
 recvLoop:
 	for {
-		msg, err := c.receiveWithRetry(ctx, &consecutiveErrors)
-		if err != nil {
-			subscriptionErr = err
-			break recvLoop
-		}
-
 		select {
 		case sem <- struct{}{}:
 		case <-ctx.Done():
 			break recvLoop
 		}
 
+		msg, err := c.receiveWithRetry(ctx, &consecutiveErrors)
+		if err != nil {
+			<-sem
+			subscriptionErr = err
+			break recvLoop
+		}
+
 		go func() {
 			defer func() { <-sem }() // Release the semaphore.
 
diff --git a/internal/consumer/consumer_test.go b/internal/consumer/consumer_test.go
index c6773d923..23756eb31 100644
--- a/internal/consumer/consumer_test.go
+++ b/internal/consumer/consumer_test.go
@@ -3,7 +3,9 @@ package consumer_test
 import (
 	"context"
 	"log"
+	"sync/atomic"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/hookdeck/outpost/internal/consumer"
@@ -278,3 +280,42 @@ var _ consumer.MessageHandler = &handlerImpl{}
 func (h *handlerImpl) Handle(ctx context.Context, msg *mqs.Message) error {
 	return h.handle(ctx, msg)
 }
+
+// With every handler busy, the consumer must not receive another message: it
+// would wait for a slot with its visibility timeout running, and be dropped
+// unacked on shutdown.
+func TestConsumer_ReceivesOnlyWithFreeSlot(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		ctx, cancel := context.WithCancel(t.Context())
+
+		var receives atomic.Int32
+		sub := &fakeSubscription{
+			receive: func(ctx context.Context) (*mqs.Message, error) {
+				receives.Add(1)
+				return &mqs.Message{Body: []byte("m")}, nil
+			},
+		}
+		release := make(chan struct{})
+		handler := &handlerImpl{
+			handle: func(ctx context.Context, msg *mqs.Message) error {
+				<-release
+				return nil
+			},
+		}
+
+		done := make(chan error)
+		go func() { done <- consumer.New(sub, handler, consumer.WithConcurrency(2)).Run(ctx) }()
+		synctest.Wait()
+		assert.Equal(t, int32(2), receives.Load(), "one receive per free slot")
+
+		release <- struct{}{}
+		synctest.Wait()
+		assert.Equal(t, int32(3), receives.Load(), "a freed slot allows one more receive")
+
+		cancel()
+		close(release)
+		assert.NoError(t, <-done)
+		assert.Equal(t, int32(3), receives.Load(), "nothing received after shutdown")
+	})
+}

From 303d95a2dfe43b881b803baa83bf1cd08649ced8 Mon Sep 17 00:00:00 2001
From: Alex Luong <alex@alexluong.com>
Date: Tue, 27 Oct 2026 20:18:13 +0700
Subject: [PATCH 2/2] docs: clarify consumer slot comments

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 internal/consumer/consumer.go      | 4 ++--
 internal/consumer/consumer_test.go | 4 +---
 2 files changed, 3 insertions(+), 5 deletions(-)

diff --git a/internal/consumer/consumer.go b/internal/consumer/consumer.go
index 233ad099a..88ae396a6 100644
--- a/internal/consumer/consumer.go
+++ b/internal/consumer/consumer.go
@@ -206,8 +206,8 @@ recvLoop:
 }
 
 // runWithSemaphore limits concurrency via a channel-based semaphore. The slot
-// is taken before receiving, so a message is only received when a handler can
-// run it now.
+// is taken before receiving: a received message must not wait for a handler
+// while its visibility timeout runs.
 func (c *consumerImpl) runWithSemaphore(ctx context.Context) error {
 	tracer := otel.GetTracerProvider().Tracer("github.com/hookdeck/outpost/internal/consumer")
 
diff --git a/internal/consumer/consumer_test.go b/internal/consumer/consumer_test.go
index 23756eb31..9e30482bf 100644
--- a/internal/consumer/consumer_test.go
+++ b/internal/consumer/consumer_test.go
@@ -281,9 +281,7 @@ func (h *handlerImpl) Handle(ctx context.Context, msg *mqs.Message) error {
 	return h.handle(ctx, msg)
 }
 
-// With every handler busy, the consumer must not receive another message: it
-// would wait for a slot with its visibility timeout running, and be dropped
-// unacked on shutdown.
+// The consumer receives only when a handler slot is free.
 func TestConsumer_ReceivesOnlyWithFreeSlot(t *testing.T) {
 	t.Parallel()
 	synctest.Test(t, func(t *testing.T) {