From 802a7a6d258f04eee636b4837554a043badde066 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Wed, 17 Jun 2026 15:54:48 -0700
Subject: [PATCH 01/15] Add processor graceful shutdown

---
 v2/inflightmessages.go      |  62 ++++++++++++
 v2/inflightmessages_test.go | 167 +++++++++++++++++++++++++++++++
 v2/processor.go             |  84 ++++++++++++----
 v2/processor_fake_test.go   |  58 +++++++++--
 v2/processor_test.go        | 190 ++++++++++++++++++++++++++++++++++++
 5 files changed, 538 insertions(+), 23 deletions(-)
 create mode 100644 v2/inflightmessages.go
 create mode 100644 v2/inflightmessages_test.go

diff --git a/v2/inflightmessages.go b/v2/inflightmessages.go
new file mode 100644
index 00000000..984682f9
--- /dev/null
+++ b/v2/inflightmessages.go
@@ -0,0 +1,62 @@
+package shuttle
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
+)
+
+const inFlightMessageAbandonTimeout = 10 * time.Second
+
+type inFlightMessages struct {
+	mu      sync.RWMutex
+	tracked map[*azservicebus.ReceivedMessage]struct{}
+}
+
+func newInFlightMessages() *inFlightMessages {
+	return &inFlightMessages{
+		tracked: make(map[*azservicebus.ReceivedMessage]struct{}),
+	}
+}
+
+func (m *inFlightMessages) track(message *azservicebus.ReceivedMessage) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.tracked[message] = struct{}{}
+}
+
+func (m *inFlightMessages) forget(message *azservicebus.ReceivedMessage) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	delete(m.tracked, message)
+}
+
+func (m *inFlightMessages) close(ctx context.Context, settler MessageSettler) error {
+	var errs []error
+	for _, message := range m.messages() {
+		abandonCtx, cancel := context.WithTimeout(ctx, inFlightMessageAbandonTimeout)
+		err := settler.AbandonMessage(abandonCtx, message, nil)
+		cancel()
+		if err != nil {
+			errs = append(errs, fmt.Errorf("failed to abandon message %s during processor close: %w", message.MessageID, err))
+			continue
+		}
+		m.forget(message)
+	}
+	return errors.Join(errs...)
+}
+
+func (m *inFlightMessages) messages() []*azservicebus.ReceivedMessage {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	messages := make([]*azservicebus.ReceivedMessage, 0, len(m.tracked))
+	for message := range m.tracked {
+		messages = append(messages, message)
+	}
+	return messages
+}
diff --git a/v2/inflightmessages_test.go b/v2/inflightmessages_test.go
new file mode 100644
index 00000000..e84cc459
--- /dev/null
+++ b/v2/inflightmessages_test.go
@@ -0,0 +1,167 @@
+package shuttle
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
+	"github.com/stretchr/testify/require"
+)
+
+func TestInFlightMessages_CloseAbandonsTrackedMessages(t *testing.T) {
+	inFlight := newInFlightMessages()
+	settler := &inFlightMessageSettler{}
+	trackedMessage := &azservicebus.ReceivedMessage{MessageID: "tracked"}
+	forgottenMessage := &azservicebus.ReceivedMessage{MessageID: "forgotten"}
+
+	inFlight.track(trackedMessage)
+	inFlight.track(forgottenMessage)
+	inFlight.forget(forgottenMessage)
+
+	require.NoError(t, inFlight.close(context.Background(), settler))
+
+	messages := settler.abandonedMessages()
+	require.Len(t, messages, 1)
+	require.Same(t, trackedMessage, messages[0])
+}
+
+func TestInFlightMessages_CloseRemovesTrackedMessages(t *testing.T) {
+	inFlight := newInFlightMessages()
+	settler := &inFlightMessageSettler{}
+	message := &azservicebus.ReceivedMessage{MessageID: "message"}
+
+	inFlight.track(message)
+
+	require.NoError(t, inFlight.close(context.Background(), settler))
+	require.NoError(t, inFlight.close(context.Background(), settler))
+
+	messages := settler.abandonedMessages()
+	require.Len(t, messages, 1)
+	require.Same(t, message, messages[0])
+}
+
+func TestInFlightMessages_CloseReturnsAbandonErrors(t *testing.T) {
+	firstErr := errors.New("first abandon failed")
+	secondErr := errors.New("second abandon failed")
+	inFlight := newInFlightMessages()
+	settler := &inFlightMessageSettler{
+		abandonErrors: []error{firstErr, secondErr},
+	}
+
+	inFlight.track(&azservicebus.ReceivedMessage{MessageID: "first"})
+	inFlight.track(&azservicebus.ReceivedMessage{MessageID: "second"})
+
+	err := inFlight.close(context.Background(), settler)
+
+	require.Error(t, err)
+	require.ErrorIs(t, err, firstErr)
+	require.ErrorIs(t, err, secondErr)
+}
+
+func TestInFlightMessages_CloseKeepsMessagesWhenAbandonFails(t *testing.T) {
+	abandonErr := errors.New("abandon failed")
+	inFlight := newInFlightMessages()
+	failingSettler := &inFlightMessageSettler{
+		abandonErrors: []error{abandonErr},
+	}
+	message := &azservicebus.ReceivedMessage{MessageID: "message"}
+
+	inFlight.track(message)
+
+	err := inFlight.close(context.Background(), failingSettler)
+
+	require.ErrorIs(t, err, abandonErr)
+	messages := inFlight.messages()
+	require.Len(t, messages, 1)
+	require.Same(t, message, messages[0])
+
+	successfulSettler := &inFlightMessageSettler{}
+	require.NoError(t, inFlight.close(context.Background(), successfulSettler))
+	require.Empty(t, inFlight.messages())
+	abandonedMessages := successfulSettler.abandonedMessages()
+	require.Len(t, abandonedMessages, 1)
+	require.Same(t, message, abandonedMessages[0])
+}
+
+func TestInFlightMessages_CloseUsesAbandonTimeout(t *testing.T) {
+	inFlight := newInFlightMessages()
+	settler := &inFlightMessageSettler{}
+
+	inFlight.track(&azservicebus.ReceivedMessage{MessageID: "message"})
+
+	beforeClose := time.Now()
+	require.NoError(t, inFlight.close(context.Background(), settler))
+	afterClose := time.Now()
+
+	deadlines := settler.abandonDeadlines()
+	require.Len(t, deadlines, 1)
+	require.True(t, deadlines[0].ok)
+	require.True(t, deadlines[0].deadline.After(beforeClose.Add(9*time.Second)))
+	require.True(t, deadlines[0].deadline.Before(afterClose.Add(11*time.Second)))
+}
+
+type inFlightMessageSettler struct {
+	mu             sync.Mutex
+	abandoned      []*azservicebus.ReceivedMessage
+	deadlines      []abandonDeadline
+	abandonErrors  []error
+	abandonAttempt int
+}
+
+type abandonDeadline struct {
+	deadline time.Time
+	ok       bool
+}
+
+func (s *inFlightMessageSettler) AbandonMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.AbandonMessageOptions) error {
+	deadline, ok := ctx.Deadline()
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.abandoned = append(s.abandoned, message)
+	s.deadlines = append(s.deadlines, abandonDeadline{
+		deadline: deadline,
+		ok:       ok,
+	})
+	err := error(nil)
+	if s.abandonAttempt < len(s.abandonErrors) {
+		err = s.abandonErrors[s.abandonAttempt]
+	}
+	s.abandonAttempt++
+	return err
+}
+
+func (s *inFlightMessageSettler) CompleteMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.CompleteMessageOptions) error {
+	return nil
+}
+
+func (s *inFlightMessageSettler) DeadLetterMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.DeadLetterOptions) error {
+	return nil
+}
+
+func (s *inFlightMessageSettler) DeferMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.DeferMessageOptions) error {
+	return nil
+}
+
+func (s *inFlightMessageSettler) RenewMessageLock(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.RenewMessageLockOptions) error {
+	return nil
+}
+
+func (s *inFlightMessageSettler) abandonedMessages() []*azservicebus.ReceivedMessage {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	messages := make([]*azservicebus.ReceivedMessage, len(s.abandoned))
+	copy(messages, s.abandoned)
+	return messages
+}
+
+func (s *inFlightMessageSettler) abandonDeadlines() []abandonDeadline {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	deadlines := make([]abandonDeadline, len(s.deadlines))
+	copy(deadlines, s.deadlines)
+	return deadlines
+}
diff --git a/v2/processor.go b/v2/processor.go
index 814de7b5..117bd562 100644
--- a/v2/processor.go
+++ b/v2/processor.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
@@ -42,7 +43,11 @@ type Processor struct {
 	receiver          Receiver
 	options           ProcessorOptions
 	handle            Handler
-	concurrencyTokens chan struct{} // tracks how many concurrent messages are currently being handled by the processor
+	concurrencyTokens chan struct{} // TODO: remove once receive sizing and in-flight tracking share a simpler lifecycle model.
+	inFlightMessages  *inFlightMessages
+	shutdownCtx       context.Context
+	shutdownCancel    context.CancelFunc
+	receiveMu         sync.Mutex
 }
 
 // ProcessorOptions configures the processor
@@ -101,11 +106,15 @@ func applyProcessorOptions(options *ProcessorOptions) *ProcessorOptions {
 // NewProcessor creates a new processor with the provided receiver and handler.
 func NewProcessor(receiver Receiver, handler HandlerFunc, options *ProcessorOptions) *Processor {
 	opts := applyProcessorOptions(options)
+	shutdownCtx, shutdownCancel := context.WithCancel(context.Background())
 	return &Processor{
 		receiver:          receiver,
 		handle:            handler,
 		options:           *opts,
 		concurrencyTokens: make(chan struct{}, opts.MaxConcurrency),
+		inFlightMessages:  newInFlightMessages(),
+		shutdownCtx:       shutdownCtx,
+		shutdownCancel:    shutdownCancel,
 	}
 }
 
@@ -113,6 +122,11 @@ func NewProcessor(receiver Receiver, handler HandlerFunc, options *ProcessorOpti
 // It will retry starting the processor based on the StartMaxAttempt and StartRetryDelayStrategy.
 // Returns a combined list of errors encountered during each processor start attempt.
 func (p *Processor) Start(ctx context.Context) (err error) {
+	if p.isClosed() {
+		return context.Canceled
+	}
+	ctx, cancel := p.startContext(ctx)
+	defer cancel()
 	defer func() {
 		if rec := recover(); rec != nil {
 			err = fmt.Errorf("panic recovered from processor: %s", rec)
@@ -121,6 +135,15 @@ func (p *Processor) Start(ctx context.Context) (err error) {
 	return p.startWithRetries(ctx)
 }
 
+// Close stops receiving new messages, cancels in-flight message handlers, and
+// abandons messages currently held by the processor.
+func (p *Processor) Close(ctx context.Context) error {
+	p.shutdownCancel()
+	p.receiveMu.Lock()
+	defer p.receiveMu.Unlock()
+	return p.inFlightMessages.close(ctx, p.receiver)
+}
+
 // startWithRetries starts a processor and blocks until an error occurs or the context is canceled.
 // It will retry starting the processor based on the StartMaxAttempt and StartRetryDelayStrategy.
 // Returns a combined list of errors during the start attempts or ctx.Err() if the context
@@ -147,18 +170,32 @@ func (p *Processor) startWithRetries(ctx context.Context) error {
 	return savedError
 }
 
+func (p *Processor) startContext(ctx context.Context) (context.Context, context.CancelFunc) {
+	ctx, cancel := context.WithCancel(ctx)
+	if p.isClosed() {
+		cancel()
+		return ctx, cancel
+	}
+	go func() {
+		select {
+		case <-p.shutdownCtx.Done():
+			cancel()
+		case <-ctx.Done():
+		}
+	}()
+	return ctx, cancel
+}
+
+func (p *Processor) isClosed() bool {
+	return p.shutdownCtx.Err() != nil
+}
+
 // start starts the processor and blocks until an error occurs or the context is canceled.
 func (p *Processor) start(ctx context.Context) error {
 	logger := getLogger(ctx)
 	logger.Info("starting processor")
-	messages, err := p.receiver.ReceiveMessages(ctx, p.options.MaxReceiveCount, nil)
-	if err != nil {
-		return fmt.Errorf("failed to receive messages: %w", err)
-	}
-	logger.Info(fmt.Sprintf("received %d messages - initial", len(messages)))
-	processor.Metric.IncMessageReceived(float64(len(messages)))
-	for _, msg := range messages {
-		p.process(ctx, msg)
+	if err := p.receiveAndProcess(ctx, p.options.MaxReceiveCount, "initial"); err != nil {
+		return err
 	}
 	for ctx.Err() == nil {
 		select {
@@ -167,14 +204,8 @@ func (p *Processor) start(ctx context.Context) error {
 			if ctx.Err() != nil || maxMessages == 0 {
 				break
 			}
-			messages, err := p.receiver.ReceiveMessages(ctx, maxMessages, nil)
-			if err != nil {
-				return fmt.Errorf("failed to receive messages: %w", err)
-			}
-			logger.Info(fmt.Sprintf("received %d messages from processor loop", len(messages)))
-			processor.Metric.IncMessageReceived(float64(len(messages)))
-			for _, msg := range messages {
-				p.process(ctx, msg)
+			if err := p.receiveAndProcess(ctx, maxMessages, "from processor loop"); err != nil {
+				return err
 			}
 		case <-ctx.Done():
 			logger.Info("context done, stop receiving from processor")
@@ -184,13 +215,32 @@ func (p *Processor) start(ctx context.Context) error {
 	return ctx.Err()
 }
 
+func (p *Processor) receiveAndProcess(ctx context.Context, maxMessages int, source string) error {
+	p.receiveMu.Lock()
+	defer p.receiveMu.Unlock()
+
+	messages, err := p.receiver.ReceiveMessages(ctx, maxMessages, nil)
+	if err != nil {
+		return fmt.Errorf("failed to receive messages: %w", err)
+	}
+	getLogger(ctx).Info(fmt.Sprintf("received %d messages - %s", len(messages), source))
+	processor.Metric.IncMessageReceived(float64(len(messages)))
+	for _, msg := range messages {
+		p.process(ctx, msg)
+	}
+	return nil
+}
+
 func (p *Processor) process(ctx context.Context, message *azservicebus.ReceivedMessage) {
 	p.concurrencyTokens <- struct{}{}
+	p.inFlightMessages.track(message)
+
 	go func() {
 		msgContext, cancel := context.WithCancel(ctx)
 		// cancel messageContext when we get out of this goroutine
 		defer cancel()
 		defer func() {
+			p.inFlightMessages.forget(message)
 			<-p.concurrencyTokens
 			processor.Metric.IncMessageHandled(message)
 			processor.Metric.DecConcurrentMessageCount(message)
diff --git a/v2/processor_fake_test.go b/v2/processor_fake_test.go
index 4c907981..49e53eca 100644
--- a/v2/processor_fake_test.go
+++ b/v2/processor_fake_test.go
@@ -3,6 +3,7 @@ package shuttle_test
 import (
 	"context"
 	"fmt"
+	"sync"
 	"sync/atomic"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
@@ -18,11 +19,18 @@ type fakeSettler struct {
 	DeadLetterCalled atomic.Int32
 	DeferCalled      atomic.Int32
 	RenewCalled      atomic.Int32
+	SetupAbandonErr  error
+
+	mu                sync.Mutex
+	AbandonedMessages []*azservicebus.ReceivedMessage
 }
 
 func (f *fakeSettler) AbandonMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.AbandonMessageOptions) error {
 	f.AbandonCalled.Add(1)
-	return nil
+	f.mu.Lock()
+	f.AbandonedMessages = append(f.AbandonedMessages, message)
+	f.mu.Unlock()
+	return f.SetupAbandonErr
 }
 
 func (f *fakeSettler) CompleteMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.CompleteMessageOptions) error {
@@ -45,6 +53,14 @@ func (f *fakeSettler) RenewMessageLock(ctx context.Context, message *azservicebu
 	return nil
 }
 
+func (f *fakeSettler) abandonedMessages() []*azservicebus.ReceivedMessage {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	messages := make([]*azservicebus.ReceivedMessage, len(f.AbandonedMessages))
+	copy(messages, f.AbandonedMessages)
+	return messages
+}
+
 type fakeReceiver struct {
 	// outcomes to verify
 	ReceiveCalls []int // array of maxMessage value passed to receive calls in the lifetime of the fake receiver
@@ -55,21 +71,51 @@ type fakeReceiver struct {
 	*fakeSettler
 	SetupMaxReceiveCalls int
 	SetupReceivePanic    string
+	SetupRespectContext  bool
+	SetupReceiveStarted  chan struct{}
 }
 
-func (f *fakeReceiver) ReceiveMessages(_ context.Context, maxMessages int, _ *azservicebus.ReceiveMessagesOptions) ([]*azservicebus.ReceivedMessage, error) {
+func (f *fakeReceiver) ReceiveMessages(ctx context.Context, maxMessages int, _ *azservicebus.ReceiveMessagesOptions) ([]*azservicebus.ReceivedMessage, error) {
 	f.ReceiveCalls = append(f.ReceiveCalls, maxMessages)
+	if f.SetupReceiveStarted != nil {
+		select {
+		case f.SetupReceiveStarted <- struct{}{}:
+		default:
+		}
+	}
 	if maxMessages == 0 && len(f.SetupReceivedMessages) > 0 {
 		return nil, nil
 	}
 	var result []*azservicebus.ReceivedMessage
-	for msg := range f.SetupReceivedMessages {
-		result = append(result, msg)
-		if len(result) == maxMessages || len(f.SetupReceivedMessages) == 0 {
-			break
+	for len(result) < maxMessages {
+		if f.SetupRespectContext {
+			select {
+			case msg, ok := <-f.SetupReceivedMessages:
+				if !ok {
+					return f.receiveResult(result)
+				}
+				result = append(result, msg)
+				if len(f.SetupReceivedMessages) == 0 {
+					return f.receiveResult(result)
+				}
+			case <-ctx.Done():
+				return result, ctx.Err()
+			}
+			continue
+		}
+		for msg := range f.SetupReceivedMessages {
+			result = append(result, msg)
+			if len(result) == maxMessages || len(f.SetupReceivedMessages) == 0 {
+				break
+			}
 		}
+		break
 	}
 
+	return f.receiveResult(result)
+}
+
+func (f *fakeReceiver) receiveResult(result []*azservicebus.ReceivedMessage) ([]*azservicebus.ReceivedMessage, error) {
 	if f.SetupReceivePanic != "" {
 		panic(f.SetupReceivePanic)
 	}
diff --git a/v2/processor_test.go b/v2/processor_test.go
index 9e2c3d3a..49c51480 100644
--- a/v2/processor_test.go
+++ b/v2/processor_test.go
@@ -2,6 +2,7 @@ package shuttle_test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -377,6 +378,195 @@ func TestProcessorStart_RecoversReceiverPanic(t *testing.T) {
 	g.Expect(err.Error()).To(ContainSubstring("panic recovered from processor: receive panic!"))
 }
 
+func TestProcessorClose_CancelsAndAbandonsInflightMessages(t *testing.T) {
+	messages := messagesChannel(2)
+	close(messages)
+	settler := &fakeSettler{}
+	rcv := &fakeReceiver{
+		fakeSettler:           settler,
+		SetupReceivedMessages: messages,
+		SetupMaxReceiveCalls:  10,
+	}
+	started := make(chan *azservicebus.ReceivedMessage, 2)
+	canceled := make(chan *azservicebus.ReceivedMessage, 2)
+	processor := shuttle.NewProcessor(rcv, func(ctx context.Context, settler shuttle.MessageSettler, message *azservicebus.ReceivedMessage) {
+		started <- message
+		<-ctx.Done()
+		canceled <- message
+	}, &shuttle.ProcessorOptions{
+		MaxConcurrency:  2,
+		ReceiveInterval: to.Ptr(1 * time.Hour),
+	})
+
+	errCh := make(chan error, 1)
+	go func() { errCh <- processor.Start(context.Background()) }()
+
+	g := NewWithT(t)
+	g.Eventually(started).Should(Receive())
+	g.Eventually(started).Should(Receive())
+
+	g.Expect(processor.Close(context.Background())).To(Succeed())
+	g.Eventually(canceled).Should(Receive())
+	g.Eventually(canceled).Should(Receive())
+	g.Eventually(errCh).Should(Receive(MatchError(context.Canceled)))
+	g.Expect(settler.AbandonCalled.Load()).To(Equal(int32(2)))
+	g.Expect(settler.abandonedMessages()).To(HaveLen(2))
+}
+
+func TestProcessorClose_StopsStartReceiveLoop(t *testing.T) {
+	rcv := &fakeReceiver{
+		fakeSettler:           &fakeSettler{},
+		SetupReceivedMessages: make(chan *azservicebus.ReceivedMessage),
+		SetupMaxReceiveCalls:  10,
+		SetupRespectContext:   true,
+		SetupReceiveStarted:   make(chan struct{}, 1),
+	}
+	processor := shuttle.NewProcessor(rcv, MyHandler(0*time.Second), &shuttle.ProcessorOptions{
+		ReceiveInterval: to.Ptr(10 * time.Millisecond),
+	})
+	errCh := make(chan error, 1)
+	go func() { errCh <- processor.Start(context.Background()) }()
+
+	g := NewWithT(t)
+	g.Eventually(rcv.SetupReceiveStarted).Should(Receive())
+	g.Expect(processor.Close(context.Background())).To(Succeed())
+	g.Eventually(errCh).Should(Receive(MatchError(MatchRegexp("failed to receive messages: context canceled"))))
+	g.Expect(rcv.ReceiveCalls).To(HaveLen(1))
+}
+
+func TestProcessorClose_WaitsForReceiveBatchBeforeAbandoning(t *testing.T) {
+	messages := make(chan *azservicebus.ReceivedMessage)
+	settler := &fakeSettler{}
+	rcv := &fakeReceiver{
+		fakeSettler:           settler,
+		SetupReceivedMessages: messages,
+		SetupMaxReceiveCalls:  10,
+		SetupReceiveStarted:   make(chan struct{}, 1),
+	}
+	releaseHandler := make(chan struct{})
+	processor := shuttle.NewProcessor(rcv, func(ctx context.Context, settler shuttle.MessageSettler, message *azservicebus.ReceivedMessage) {
+		<-releaseHandler
+	}, &shuttle.ProcessorOptions{
+		ReceiveInterval: to.Ptr(1 * time.Hour),
+	})
+	errCh := make(chan error, 1)
+	go func() { errCh <- processor.Start(context.Background()) }()
+
+	g := NewWithT(t)
+	g.Eventually(rcv.SetupReceiveStarted).Should(Receive())
+
+	closeErrCh := make(chan error, 1)
+	go func() { closeErrCh <- processor.Close(context.Background()) }()
+	g.Consistently(closeErrCh, 20*time.Millisecond).ShouldNot(Receive())
+
+	message := &azservicebus.ReceivedMessage{MessageID: "received-during-close"}
+	messages <- message
+
+	g.Eventually(closeErrCh).Should(Receive(Succeed()))
+	abandoned := settler.abandonedMessages()
+	g.Expect(abandoned).To(HaveLen(1))
+	g.Expect(abandoned[0]).To(BeIdenticalTo(message))
+
+	close(releaseHandler)
+	g.Eventually(errCh).Should(Receive(MatchError(context.Canceled)))
+}
+
+func TestProcessorClose_ReturnsAbandonErrors(t *testing.T) {
+	abandonErr := errors.New("abandon failed")
+	messages := messagesChannel(2)
+	close(messages)
+	settler := &fakeSettler{SetupAbandonErr: abandonErr}
+	rcv := &fakeReceiver{
+		fakeSettler:           settler,
+		SetupReceivedMessages: messages,
+		SetupMaxReceiveCalls:  10,
+	}
+	started := make(chan struct{}, 2)
+	release := make(chan struct{})
+	processor := shuttle.NewProcessor(rcv, func(ctx context.Context, settler shuttle.MessageSettler, message *azservicebus.ReceivedMessage) {
+		started <- struct{}{}
+		<-release
+	}, &shuttle.ProcessorOptions{
+		MaxConcurrency:  2,
+		ReceiveInterval: to.Ptr(1 * time.Hour),
+	})
+	errCh := make(chan error, 1)
+	go func() { errCh <- processor.Start(context.Background()) }()
+	defer func() {
+		close(release)
+		<-errCh
+	}()
+
+	g := NewWithT(t)
+	g.Eventually(started).Should(Receive())
+	g.Eventually(started).Should(Receive())
+
+	err := processor.Close(context.Background())
+	g.Expect(err).To(HaveOccurred())
+	g.Expect(errors.Is(err, abandonErr)).To(BeTrue())
+	g.Expect(err.Error()).To(ContainSubstring("failed to abandon message"))
+	g.Expect(settler.AbandonCalled.Load()).To(Equal(int32(2)))
+}
+
+func TestProcessorClose_IsIdempotent(t *testing.T) {
+	messages := messagesChannel(1)
+	close(messages)
+	settler := &fakeSettler{}
+	rcv := &fakeReceiver{
+		fakeSettler:           settler,
+		SetupReceivedMessages: messages,
+		SetupMaxReceiveCalls:  10,
+	}
+	started := make(chan struct{}, 1)
+	processor := shuttle.NewProcessor(rcv, func(ctx context.Context, settler shuttle.MessageSettler, message *azservicebus.ReceivedMessage) {
+		started <- struct{}{}
+		<-ctx.Done()
+	}, &shuttle.ProcessorOptions{
+		ReceiveInterval: to.Ptr(1 * time.Hour),
+	})
+	errCh := make(chan error, 1)
+	go func() { errCh <- processor.Start(context.Background()) }()
+
+	g := NewWithT(t)
+	g.Eventually(started).Should(Receive())
+	g.Expect(processor.Close(context.Background())).To(Succeed())
+	g.Expect(processor.Close(context.Background())).To(Succeed())
+	g.Eventually(errCh).Should(Receive(MatchError(context.Canceled)))
+	g.Expect(settler.AbandonCalled.Load()).To(Equal(int32(1)))
+}
+
+func TestProcessorClose_DoesNotWaitForHandlerToExit(t *testing.T) {
+	messages := messagesChannel(1)
+	close(messages)
+	settler := &fakeSettler{}
+	rcv := &fakeReceiver{
+		fakeSettler:           settler,
+		SetupReceivedMessages: messages,
+		SetupMaxReceiveCalls:  10,
+	}
+	started := make(chan struct{}, 1)
+	release := make(chan struct{})
+	processor := shuttle.NewProcessor(rcv, func(ctx context.Context, settler shuttle.MessageSettler, message *azservicebus.ReceivedMessage) {
+		started <- struct{}{}
+		<-release
+	}, &shuttle.ProcessorOptions{
+		ReceiveInterval: to.Ptr(1 * time.Hour),
+	})
+	errCh := make(chan error, 1)
+	go func() { errCh <- processor.Start(context.Background()) }()
+
+	g := NewWithT(t)
+	g.Eventually(started).Should(Receive())
+
+	closeCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	g.Expect(processor.Close(closeCtx)).To(Succeed())
+	g.Expect(settler.AbandonCalled.Load()).To(Equal(int32(1)))
+	g.Eventually(errCh).Should(Receive(MatchError(context.Canceled)))
+
+	close(release)
+}
+
 func messagesChannel(messageCount int) chan *azservicebus.ReceivedMessage {
 	messages := make(chan *azservicebus.ReceivedMessage, messageCount)
 	for i := 0; i < messageCount; i++ {

From 8c49ab9b965ac082582c337413c16786cc1dfb99 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 10:11:42 -0700
Subject: [PATCH 02/15] Abandon in-flight messages concurrently on close

---
 v2/inflightmessages.go      | 32 +++++++++-----
 v2/inflightmessages_test.go | 85 +++++++++++++++++++++++++++++++++++++
 2 files changed, 107 insertions(+), 10 deletions(-)

diff --git a/v2/inflightmessages.go b/v2/inflightmessages.go
index 984682f9..9e0b33e6 100644
--- a/v2/inflightmessages.go
+++ b/v2/inflightmessages.go
@@ -36,17 +36,29 @@ func (m *inFlightMessages) forget(message *azservicebus.ReceivedMessage) {
 }
 
 func (m *inFlightMessages) close(ctx context.Context, settler MessageSettler) error {
-	var errs []error
-	for _, message := range m.messages() {
-		abandonCtx, cancel := context.WithTimeout(ctx, inFlightMessageAbandonTimeout)
-		err := settler.AbandonMessage(abandonCtx, message, nil)
-		cancel()
-		if err != nil {
-			errs = append(errs, fmt.Errorf("failed to abandon message %s during processor close: %w", message.MessageID, err))
-			continue
-		}
-		m.forget(message)
+	messages := m.messages()
+	errs := make([]error, len(messages))
+	var wg sync.WaitGroup
+	wg.Add(len(messages))
+
+	for i, message := range messages {
+		i, message := i, message
+		go func() {
+			defer wg.Done()
+
+			abandonCtx, cancel := context.WithTimeout(ctx, inFlightMessageAbandonTimeout)
+			defer cancel()
+
+			err := settler.AbandonMessage(abandonCtx, message, nil)
+			if err != nil {
+				errs[i] = fmt.Errorf("failed to abandon message %s during processor close: %w", message.MessageID, err)
+				return
+			}
+			m.forget(message)
+		}()
 	}
+
+	wg.Wait()
 	return errors.Join(errs...)
 }
 
diff --git a/v2/inflightmessages_test.go b/v2/inflightmessages_test.go
index e84cc459..93db2110 100644
--- a/v2/inflightmessages_test.go
+++ b/v2/inflightmessages_test.go
@@ -28,6 +28,55 @@ func TestInFlightMessages_CloseAbandonsTrackedMessages(t *testing.T) {
 	require.Same(t, trackedMessage, messages[0])
 }
 
+func TestInFlightMessages_CloseAbandonsTrackedMessagesConcurrently(t *testing.T) {
+	inFlight := newInFlightMessages()
+	firstMessage := &azservicebus.ReceivedMessage{MessageID: "first"}
+	secondMessage := &azservicebus.ReceivedMessage{MessageID: "second"}
+	abandonStarted := make(chan *azservicebus.ReceivedMessage, 2)
+	releaseAbandons := make(chan struct{})
+	var releaseOnce sync.Once
+	settler := &blockingInFlightMessageSettler{
+		abandonStarted: abandonStarted,
+		releaseAbandon: releaseAbandons,
+	}
+
+	release := func() {
+		releaseOnce.Do(func() {
+			close(releaseAbandons)
+		})
+	}
+	t.Cleanup(release)
+
+	inFlight.track(firstMessage)
+	inFlight.track(secondMessage)
+
+	closeErr := make(chan error, 1)
+	go func() {
+		closeErr <- inFlight.close(context.Background(), settler)
+	}()
+
+	abandoned := map[*azservicebus.ReceivedMessage]struct{}{}
+	for len(abandoned) < 2 {
+		select {
+		case message := <-abandonStarted:
+			abandoned[message] = struct{}{}
+		case <-time.After(1 * time.Second):
+			t.Fatalf("timed out waiting for concurrent abandon attempts")
+		}
+	}
+
+	select {
+	case err := <-closeErr:
+		t.Fatalf("close returned before abandon attempts were released: %v", err)
+	default:
+	}
+
+	release()
+	require.NoError(t, <-closeErr)
+	require.Contains(t, abandoned, firstMessage)
+	require.Contains(t, abandoned, secondMessage)
+}
+
 func TestInFlightMessages_CloseRemovesTrackedMessages(t *testing.T) {
 	inFlight := newInFlightMessages()
 	settler := &inFlightMessageSettler{}
@@ -165,3 +214,39 @@ func (s *inFlightMessageSettler) abandonDeadlines() []abandonDeadline {
 	copy(deadlines, s.deadlines)
 	return deadlines
 }
+
+type blockingInFlightMessageSettler struct {
+	abandonStarted chan<- *azservicebus.ReceivedMessage
+	releaseAbandon <-chan struct{}
+}
+
+func (s *blockingInFlightMessageSettler) AbandonMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.AbandonMessageOptions) error {
+	select {
+	case s.abandonStarted <- message:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	select {
+	case <-s.releaseAbandon:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func (s *blockingInFlightMessageSettler) CompleteMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.CompleteMessageOptions) error {
+	return nil
+}
+
+func (s *blockingInFlightMessageSettler) DeadLetterMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.DeadLetterOptions) error {
+	return nil
+}
+
+func (s *blockingInFlightMessageSettler) DeferMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.DeferMessageOptions) error {
+	return nil
+}
+
+func (s *blockingInFlightMessageSettler) RenewMessageLock(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.RenewMessageLockOptions) error {
+	return nil
+}

From 1fb19d485a9d88ddb88f4410a88187d8ffdc62e5 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 11:40:54 -0700
Subject: [PATCH 03/15] Respect close context while abandoning messages

---
 v2/inflightmessages.go      | 58 +++++++++++++++++++++--------
 v2/inflightmessages_test.go | 73 +++++++++++++++++++++++++++++++++++++
 2 files changed, 115 insertions(+), 16 deletions(-)

diff --git a/v2/inflightmessages.go b/v2/inflightmessages.go
index 9e0b33e6..06a148c3 100644
--- a/v2/inflightmessages.go
+++ b/v2/inflightmessages.go
@@ -37,31 +37,57 @@ func (m *inFlightMessages) forget(message *azservicebus.ReceivedMessage) {
 
 func (m *inFlightMessages) close(ctx context.Context, settler MessageSettler) error {
 	messages := m.messages()
-	errs := make([]error, len(messages))
-	var wg sync.WaitGroup
-	wg.Add(len(messages))
+	abandonResults := make(chan error, len(messages))
 
-	for i, message := range messages {
-		i, message := i, message
+	for _, message := range messages {
+		message := message
 		go func() {
-			defer wg.Done()
-
-			abandonCtx, cancel := context.WithTimeout(ctx, inFlightMessageAbandonTimeout)
-			defer cancel()
+			abandonResults <- m.abandon(ctx, settler, message)
+		}()
+	}
 
-			err := settler.AbandonMessage(abandonCtx, message, nil)
+	var errs []error
+	for range messages {
+		select {
+		case err := <-abandonResults:
 			if err != nil {
-				errs[i] = fmt.Errorf("failed to abandon message %s during processor close: %w", message.MessageID, err)
-				return
+				errs = append(errs, err)
 			}
-			m.forget(message)
-		}()
+		case <-ctx.Done():
+			errs = append(errs, ctx.Err())
+			errs = append(errs, drainAbandonErrors(abandonResults)...)
+			return errors.Join(errs...)
+		}
 	}
-
-	wg.Wait()
 	return errors.Join(errs...)
 }
 
+func (m *inFlightMessages) abandon(ctx context.Context, settler MessageSettler, message *azservicebus.ReceivedMessage) error {
+	abandonCtx, cancel := context.WithTimeout(ctx, inFlightMessageAbandonTimeout)
+	defer cancel()
+
+	err := settler.AbandonMessage(abandonCtx, message, nil)
+	if err != nil {
+		return fmt.Errorf("failed to abandon message %s during processor close: %w", message.MessageID, err)
+	}
+	m.forget(message)
+	return nil
+}
+
+func drainAbandonErrors(abandonResults <-chan error) []error {
+	var errs []error
+	for {
+		select {
+		case err := <-abandonResults:
+			if err != nil {
+				errs = append(errs, err)
+			}
+		default:
+			return errs
+		}
+	}
+}
+
 func (m *inFlightMessages) messages() []*azservicebus.ReceivedMessage {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
diff --git a/v2/inflightmessages_test.go b/v2/inflightmessages_test.go
index 93db2110..0828c094 100644
--- a/v2/inflightmessages_test.go
+++ b/v2/inflightmessages_test.go
@@ -77,6 +77,52 @@ func TestInFlightMessages_CloseAbandonsTrackedMessagesConcurrently(t *testing.T)
 	require.Contains(t, abandoned, secondMessage)
 }
 
+func TestInFlightMessages_CloseReturnsWhenContextIsCanceled(t *testing.T) {
+	inFlight := newInFlightMessages()
+	abandonStarted := make(chan *azservicebus.ReceivedMessage, 1)
+	releaseAbandon := make(chan struct{})
+	var releaseOnce sync.Once
+	settler := &contextIgnoringBlockingInFlightMessageSettler{
+		abandonStarted: abandonStarted,
+		releaseAbandon: releaseAbandon,
+	}
+
+	release := func() {
+		releaseOnce.Do(func() {
+			close(releaseAbandon)
+		})
+	}
+	t.Cleanup(release)
+
+	inFlight.track(&azservicebus.ReceivedMessage{MessageID: "blocked"})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	closeErr := make(chan error, 1)
+	go func() {
+		closeErr <- inFlight.close(ctx, settler)
+	}()
+
+	select {
+	case <-abandonStarted:
+	case <-time.After(1 * time.Second):
+		t.Fatal("timed out waiting for abandon attempt to start")
+	}
+
+	cancel()
+
+	select {
+	case err := <-closeErr:
+		require.ErrorIs(t, err, context.Canceled)
+	case <-time.After(1 * time.Second):
+		t.Fatal("timed out waiting for close to return after context cancellation")
+	}
+
+	release()
+	require.Eventually(t, func() bool {
+		return len(inFlight.messages()) == 0
+	}, 1*time.Second, 10*time.Millisecond)
+}
+
 func TestInFlightMessages_CloseRemovesTrackedMessages(t *testing.T) {
 	inFlight := newInFlightMessages()
 	settler := &inFlightMessageSettler{}
@@ -250,3 +296,30 @@ func (s *blockingInFlightMessageSettler) DeferMessage(ctx context.Context, messa
 func (s *blockingInFlightMessageSettler) RenewMessageLock(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.RenewMessageLockOptions) error {
 	return nil
 }
+
+type contextIgnoringBlockingInFlightMessageSettler struct {
+	abandonStarted chan<- *azservicebus.ReceivedMessage
+	releaseAbandon <-chan struct{}
+}
+
+func (s *contextIgnoringBlockingInFlightMessageSettler) AbandonMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.AbandonMessageOptions) error {
+	s.abandonStarted <- message
+	<-s.releaseAbandon
+	return nil
+}
+
+func (s *contextIgnoringBlockingInFlightMessageSettler) CompleteMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.CompleteMessageOptions) error {
+	return nil
+}
+
+func (s *contextIgnoringBlockingInFlightMessageSettler) DeadLetterMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.DeadLetterOptions) error {
+	return nil
+}
+
+func (s *contextIgnoringBlockingInFlightMessageSettler) DeferMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.DeferMessageOptions) error {
+	return nil
+}
+
+func (s *contextIgnoringBlockingInFlightMessageSettler) RenewMessageLock(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.RenewMessageLockOptions) error {
+	return nil
+}

From e58b38997870cef70a3bc608a24846326f7e3087 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 12:26:44 -0700
Subject: [PATCH 04/15] Simplify processor shutdown flow

---
 v2/inflightmessages.go      | 26 ++---------------
 v2/inflightmessages_test.go | 13 +++++----
 v2/processor.go             | 56 +++++++++++++++++++++----------------
 3 files changed, 41 insertions(+), 54 deletions(-)

diff --git a/v2/inflightmessages.go b/v2/inflightmessages.go
index 06a148c3..8b2155f7 100644
--- a/v2/inflightmessages.go
+++ b/v2/inflightmessages.go
@@ -5,13 +5,10 @@ import (
 	"errors"
 	"fmt"
 	"sync"
-	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
 )
 
-const inFlightMessageAbandonTimeout = 10 * time.Second
-
 type inFlightMessages struct {
 	mu      sync.RWMutex
 	tracked map[*azservicebus.ReceivedMessage]struct{}
@@ -54,19 +51,14 @@ func (m *inFlightMessages) close(ctx context.Context, settler MessageSettler) er
 				errs = append(errs, err)
 			}
 		case <-ctx.Done():
-			errs = append(errs, ctx.Err())
-			errs = append(errs, drainAbandonErrors(abandonResults)...)
-			return errors.Join(errs...)
+			return errors.Join(append(errs, ctx.Err())...)
 		}
 	}
 	return errors.Join(errs...)
 }
 
 func (m *inFlightMessages) abandon(ctx context.Context, settler MessageSettler, message *azservicebus.ReceivedMessage) error {
-	abandonCtx, cancel := context.WithTimeout(ctx, inFlightMessageAbandonTimeout)
-	defer cancel()
-
-	err := settler.AbandonMessage(abandonCtx, message, nil)
+	err := settler.AbandonMessage(ctx, message, nil)
 	if err != nil {
 		return fmt.Errorf("failed to abandon message %s during processor close: %w", message.MessageID, err)
 	}
@@ -74,20 +66,6 @@ func (m *inFlightMessages) abandon(ctx context.Context, settler MessageSettler,
 	return nil
 }
 
-func drainAbandonErrors(abandonResults <-chan error) []error {
-	var errs []error
-	for {
-		select {
-		case err := <-abandonResults:
-			if err != nil {
-				errs = append(errs, err)
-			}
-		default:
-			return errs
-		}
-	}
-}
-
 func (m *inFlightMessages) messages() []*azservicebus.ReceivedMessage {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
diff --git a/v2/inflightmessages_test.go b/v2/inflightmessages_test.go
index 0828c094..531cc4b0 100644
--- a/v2/inflightmessages_test.go
+++ b/v2/inflightmessages_test.go
@@ -181,21 +181,22 @@ func TestInFlightMessages_CloseKeepsMessagesWhenAbandonFails(t *testing.T) {
 	require.Same(t, message, abandonedMessages[0])
 }
 
-func TestInFlightMessages_CloseUsesAbandonTimeout(t *testing.T) {
+func TestInFlightMessages_CloseUsesCloseContext(t *testing.T) {
 	inFlight := newInFlightMessages()
 	settler := &inFlightMessageSettler{}
 
 	inFlight.track(&azservicebus.ReceivedMessage{MessageID: "message"})
 
-	beforeClose := time.Now()
-	require.NoError(t, inFlight.close(context.Background(), settler))
-	afterClose := time.Now()
+	deadline := time.Now().Add(1 * time.Minute)
+	ctx, cancel := context.WithDeadline(context.Background(), deadline)
+	defer cancel()
+
+	require.NoError(t, inFlight.close(ctx, settler))
 
 	deadlines := settler.abandonDeadlines()
 	require.Len(t, deadlines, 1)
 	require.True(t, deadlines[0].ok)
-	require.True(t, deadlines[0].deadline.After(beforeClose.Add(9*time.Second)))
-	require.True(t, deadlines[0].deadline.Before(afterClose.Add(11*time.Second)))
+	require.True(t, deadlines[0].deadline.Equal(deadline))
 }
 
 type inFlightMessageSettler struct {
diff --git a/v2/processor.go b/v2/processor.go
index 117bd562..fe85ba2a 100644
--- a/v2/processor.go
+++ b/v2/processor.go
@@ -43,10 +43,11 @@ type Processor struct {
 	receiver          Receiver
 	options           ProcessorOptions
 	handle            Handler
-	concurrencyTokens chan struct{} // TODO: remove once receive sizing and in-flight tracking share a simpler lifecycle model.
+	concurrencyTokens chan struct{} // tracks how many concurrent messages are currently being handled by the processor
 	inFlightMessages  *inFlightMessages
-	shutdownCtx       context.Context
-	shutdownCancel    context.CancelFunc
+	lifecycleMu       sync.Mutex
+	closed            bool
+	cancelStart       context.CancelFunc
 	receiveMu         sync.Mutex
 }
 
@@ -106,15 +107,12 @@ func applyProcessorOptions(options *ProcessorOptions) *ProcessorOptions {
 // NewProcessor creates a new processor with the provided receiver and handler.
 func NewProcessor(receiver Receiver, handler HandlerFunc, options *ProcessorOptions) *Processor {
 	opts := applyProcessorOptions(options)
-	shutdownCtx, shutdownCancel := context.WithCancel(context.Background())
 	return &Processor{
 		receiver:          receiver,
 		handle:            handler,
 		options:           *opts,
 		concurrencyTokens: make(chan struct{}, opts.MaxConcurrency),
 		inFlightMessages:  newInFlightMessages(),
-		shutdownCtx:       shutdownCtx,
-		shutdownCancel:    shutdownCancel,
 	}
 }
 
@@ -122,10 +120,11 @@ func NewProcessor(receiver Receiver, handler HandlerFunc, options *ProcessorOpti
 // It will retry starting the processor based on the StartMaxAttempt and StartRetryDelayStrategy.
 // Returns a combined list of errors encountered during each processor start attempt.
 func (p *Processor) Start(ctx context.Context) (err error) {
-	if p.isClosed() {
+	ctx, cancel, ok := p.startContext(ctx)
+	if !ok {
 		return context.Canceled
 	}
-	ctx, cancel := p.startContext(ctx)
+	defer p.clearStartCancel()
 	defer cancel()
 	defer func() {
 		if rec := recover(); rec != nil {
@@ -138,7 +137,16 @@ func (p *Processor) Start(ctx context.Context) (err error) {
 // Close stops receiving new messages, cancels in-flight message handlers, and
 // abandons messages currently held by the processor.
 func (p *Processor) Close(ctx context.Context) error {
-	p.shutdownCancel()
+	p.lifecycleMu.Lock()
+	p.closed = true
+	cancelStart := p.cancelStart
+	p.cancelStart = nil
+	p.lifecycleMu.Unlock()
+
+	if cancelStart != nil {
+		cancelStart()
+	}
+
 	p.receiveMu.Lock()
 	defer p.receiveMu.Unlock()
 	return p.inFlightMessages.close(ctx, p.receiver)
@@ -170,24 +178,24 @@ func (p *Processor) startWithRetries(ctx context.Context) error {
 	return savedError
 }
 
-func (p *Processor) startContext(ctx context.Context) (context.Context, context.CancelFunc) {
-	ctx, cancel := context.WithCancel(ctx)
-	if p.isClosed() {
-		cancel()
-		return ctx, cancel
+func (p *Processor) startContext(ctx context.Context) (context.Context, context.CancelFunc, bool) {
+	p.lifecycleMu.Lock()
+	defer p.lifecycleMu.Unlock()
+
+	if p.closed {
+		return nil, nil, false
 	}
-	go func() {
-		select {
-		case <-p.shutdownCtx.Done():
-			cancel()
-		case <-ctx.Done():
-		}
-	}()
-	return ctx, cancel
+
+	ctx, cancel := context.WithCancel(ctx)
+	p.cancelStart = cancel
+	return ctx, cancel, true
 }
 
-func (p *Processor) isClosed() bool {
-	return p.shutdownCtx.Err() != nil
+func (p *Processor) clearStartCancel() {
+	p.lifecycleMu.Lock()
+	defer p.lifecycleMu.Unlock()
+
+	p.cancelStart = nil
 }
 
 // start starts the processor and blocks until an error occurs or the context is canceled.

From 0a5174aa9eea43a759c8e3a952462cafba8ce113 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 12:30:51 -0700
Subject: [PATCH 05/15] Move in-flight message snapshot helper

---
 v2/inflightmessages.go | 22 +++++++++++-----------
 1 file changed, 11 insertions(+), 11 deletions(-)

diff --git a/v2/inflightmessages.go b/v2/inflightmessages.go
index 8b2155f7..048c091e 100644
--- a/v2/inflightmessages.go
+++ b/v2/inflightmessages.go
@@ -32,6 +32,17 @@ func (m *inFlightMessages) forget(message *azservicebus.ReceivedMessage) {
 	delete(m.tracked, message)
 }
 
+func (m *inFlightMessages) messages() []*azservicebus.ReceivedMessage {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	messages := make([]*azservicebus.ReceivedMessage, 0, len(m.tracked))
+	for message := range m.tracked {
+		messages = append(messages, message)
+	}
+	return messages
+}
+
 func (m *inFlightMessages) close(ctx context.Context, settler MessageSettler) error {
 	messages := m.messages()
 	abandonResults := make(chan error, len(messages))
@@ -65,14 +76,3 @@ func (m *inFlightMessages) abandon(ctx context.Context, settler MessageSettler,
 	m.forget(message)
 	return nil
 }
-
-func (m *inFlightMessages) messages() []*azservicebus.ReceivedMessage {
-	m.mu.RLock()
-	defer m.mu.RUnlock()
-
-	messages := make([]*azservicebus.ReceivedMessage, 0, len(m.tracked))
-	for message := range m.tracked {
-		messages = append(messages, message)
-	}
-	return messages
-}

From 0254545009d117844955d9791fcb7f503c465816 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 13:15:05 -0700
Subject: [PATCH 06/15] Extract processor lifecycle state

---
 v2/inflightmessages.go |  8 ++++-
 v2/processor.go        | 82 ++++++++++++++++++++++++------------------
 2 files changed, 54 insertions(+), 36 deletions(-)

diff --git a/v2/inflightmessages.go b/v2/inflightmessages.go
index 048c091e..3fc155cc 100644
--- a/v2/inflightmessages.go
+++ b/v2/inflightmessages.go
@@ -55,13 +55,19 @@ func (m *inFlightMessages) close(ctx context.Context, settler MessageSettler) er
 	}
 
 	var errs []error
-	for range messages {
+	for completed := 0; completed < len(messages); completed++ {
 		select {
 		case err := <-abandonResults:
 			if err != nil {
 				errs = append(errs, err)
 			}
 		case <-ctx.Done():
+			getLogger(ctx).Warn(fmt.Sprintf(
+				"processor close context done while abandoning in-flight messages: %s; completed=%d remaining=%d",
+				ctx.Err(),
+				completed,
+				len(messages)-completed,
+			))
 			return errors.Join(append(errs, ctx.Err())...)
 		}
 	}
diff --git a/v2/processor.go b/v2/processor.go
index fe85ba2a..fe9f9da6 100644
--- a/v2/processor.go
+++ b/v2/processor.go
@@ -45,12 +45,52 @@ type Processor struct {
 	handle            Handler
 	concurrencyTokens chan struct{} // tracks how many concurrent messages are currently being handled by the processor
 	inFlightMessages  *inFlightMessages
-	lifecycleMu       sync.Mutex
-	closed            bool
-	cancelStart       context.CancelFunc
+	lifecycle         processorLifecycle
 	receiveMu         sync.Mutex
 }
 
+type processorLifecycle struct {
+	mu             sync.Mutex
+	closed         bool
+	cancelStartCtx context.CancelFunc
+}
+
+func (l *processorLifecycle) start(ctx context.Context) (context.Context, context.CancelFunc) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	if l.closed {
+		return nil, nil
+	}
+
+	ctx, cancelStartCtx := context.WithCancel(ctx)
+	l.cancelStartCtx = cancelStartCtx
+	return ctx, cancelStartCtx
+}
+
+func (l *processorLifecycle) close() {
+	if cancelStartCtx := l.closeStart(); cancelStartCtx != nil {
+		cancelStartCtx()
+	}
+}
+
+func (l *processorLifecycle) closeStart() context.CancelFunc {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	l.closed = true
+	cancelStartCtx := l.cancelStartCtx
+	l.cancelStartCtx = nil
+	return cancelStartCtx
+}
+
+func (l *processorLifecycle) clearStart() {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	l.cancelStartCtx = nil
+}
+
 // ProcessorOptions configures the processor
 // MaxConcurrency defaults to 1. Not setting MaxConcurrency, or setting it to 0 or a negative value will fallback to the default.
 // MaxReceiveCount defaults to MaxConcurrency if not set. Not setting MaxReceiveCount, or setting it to 0 or a negative value will fallback to the default.
@@ -120,11 +160,11 @@ func NewProcessor(receiver Receiver, handler HandlerFunc, options *ProcessorOpti
 // It will retry starting the processor based on the StartMaxAttempt and StartRetryDelayStrategy.
 // Returns a combined list of errors encountered during each processor start attempt.
 func (p *Processor) Start(ctx context.Context) (err error) {
-	ctx, cancel, ok := p.startContext(ctx)
-	if !ok {
+	ctx, cancel := p.lifecycle.start(ctx)
+	if ctx == nil {
 		return context.Canceled
 	}
-	defer p.clearStartCancel()
+	defer p.lifecycle.clearStart()
 	defer cancel()
 	defer func() {
 		if rec := recover(); rec != nil {
@@ -137,15 +177,7 @@ func (p *Processor) Start(ctx context.Context) (err error) {
 // Close stops receiving new messages, cancels in-flight message handlers, and
 // abandons messages currently held by the processor.
 func (p *Processor) Close(ctx context.Context) error {
-	p.lifecycleMu.Lock()
-	p.closed = true
-	cancelStart := p.cancelStart
-	p.cancelStart = nil
-	p.lifecycleMu.Unlock()
-
-	if cancelStart != nil {
-		cancelStart()
-	}
+	p.lifecycle.close()
 
 	p.receiveMu.Lock()
 	defer p.receiveMu.Unlock()
@@ -178,26 +210,6 @@ func (p *Processor) startWithRetries(ctx context.Context) error {
 	return savedError
 }
 
-func (p *Processor) startContext(ctx context.Context) (context.Context, context.CancelFunc, bool) {
-	p.lifecycleMu.Lock()
-	defer p.lifecycleMu.Unlock()
-
-	if p.closed {
-		return nil, nil, false
-	}
-
-	ctx, cancel := context.WithCancel(ctx)
-	p.cancelStart = cancel
-	return ctx, cancel, true
-}
-
-func (p *Processor) clearStartCancel() {
-	p.lifecycleMu.Lock()
-	defer p.lifecycleMu.Unlock()
-
-	p.cancelStart = nil
-}
-
 // start starts the processor and blocks until an error occurs or the context is canceled.
 func (p *Processor) start(ctx context.Context) error {
 	logger := getLogger(ctx)

From e5fce5d434ec87a35c4c657870a318fbcc54dba4 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 13:21:48 -0700
Subject: [PATCH 07/15] Return processor closed error from Start

---
 v2/processor.go      | 11 ++++++++---
 v2/processor_test.go | 18 ++++++++++++++++++
 2 files changed, 26 insertions(+), 3 deletions(-)

diff --git a/v2/processor.go b/v2/processor.go
index fe9f9da6..9b57dec5 100644
--- a/v2/processor.go
+++ b/v2/processor.go
@@ -49,6 +49,11 @@ type Processor struct {
 	receiveMu         sync.Mutex
 }
 
+// ErrProcessorClosed is returned when Start is called after the processor has been closed.
+var ErrProcessorClosed = errors.New("processor has already been closed")
+
+var errProcessorClosed = fmt.Errorf("%w: %w", ErrProcessorClosed, context.Canceled)
+
 type processorLifecycle struct {
 	mu             sync.Mutex
 	closed         bool
@@ -69,12 +74,12 @@ func (l *processorLifecycle) start(ctx context.Context) (context.Context, contex
 }
 
 func (l *processorLifecycle) close() {
-	if cancelStartCtx := l.closeStart(); cancelStartCtx != nil {
+	if cancelStartCtx := l.markClosed(); cancelStartCtx != nil {
 		cancelStartCtx()
 	}
 }
 
-func (l *processorLifecycle) closeStart() context.CancelFunc {
+func (l *processorLifecycle) markClosed() context.CancelFunc {
 	l.mu.Lock()
 	defer l.mu.Unlock()
 
@@ -162,7 +167,7 @@ func NewProcessor(receiver Receiver, handler HandlerFunc, options *ProcessorOpti
 func (p *Processor) Start(ctx context.Context) (err error) {
 	ctx, cancel := p.lifecycle.start(ctx)
 	if ctx == nil {
-		return context.Canceled
+		return errProcessorClosed
 	}
 	defer p.lifecycle.clearStart()
 	defer cancel()
diff --git a/v2/processor_test.go b/v2/processor_test.go
index 49c51480..ab560e47 100644
--- a/v2/processor_test.go
+++ b/v2/processor_test.go
@@ -101,6 +101,24 @@ func TestProcessorStart_ContextCanceledAfterStart(t *testing.T) {
 	g.Eventually(errCh).Should(Receive(MatchError(context.Canceled)))
 }
 
+func TestProcessorStart_ReturnsProcessorClosedErrorAfterClose(t *testing.T) {
+	rcv := &fakeReceiver{
+		fakeSettler:           &fakeSettler{},
+		SetupReceivedMessages: make(chan *azservicebus.ReceivedMessage),
+	}
+	close(rcv.SetupReceivedMessages)
+	processor := shuttle.NewProcessor(rcv, MyHandler(0*time.Millisecond), nil)
+
+	require.NoError(t, processor.Close(context.Background()))
+
+	err := processor.Start(context.Background())
+
+	require.ErrorIs(t, err, shuttle.ErrProcessorClosed)
+	require.ErrorIs(t, err, context.Canceled)
+	require.ErrorContains(t, err, "processor has already been closed")
+	require.Empty(t, rcv.ReceiveCalls)
+}
+
 func TestProcessorStart_CanSetMaxConcurrency(t *testing.T) {
 	a := require.New(t)
 	rcv := &fakeReceiver{

From 70f2ac741c51f952ccaa058162177844f2ae56ec Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 13:23:31 -0700
Subject: [PATCH 08/15] Remove receive mutex from processor close

---
 v2/processor.go      |  8 +-------
 v2/processor_test.go | 19 ++++---------------
 2 files changed, 5 insertions(+), 22 deletions(-)

diff --git a/v2/processor.go b/v2/processor.go
index 9b57dec5..93dfc55a 100644
--- a/v2/processor.go
+++ b/v2/processor.go
@@ -46,7 +46,6 @@ type Processor struct {
 	concurrencyTokens chan struct{} // tracks how many concurrent messages are currently being handled by the processor
 	inFlightMessages  *inFlightMessages
 	lifecycle         processorLifecycle
-	receiveMu         sync.Mutex
 }
 
 // ErrProcessorClosed is returned when Start is called after the processor has been closed.
@@ -180,12 +179,10 @@ func (p *Processor) Start(ctx context.Context) (err error) {
 }
 
 // Close stops receiving new messages, cancels in-flight message handlers, and
-// abandons messages currently held by the processor.
+// best-effort abandons messages currently held by the processor.
 func (p *Processor) Close(ctx context.Context) error {
 	p.lifecycle.close()
 
-	p.receiveMu.Lock()
-	defer p.receiveMu.Unlock()
 	return p.inFlightMessages.close(ctx, p.receiver)
 }
 
@@ -241,9 +238,6 @@ func (p *Processor) start(ctx context.Context) error {
 }
 
 func (p *Processor) receiveAndProcess(ctx context.Context, maxMessages int, source string) error {
-	p.receiveMu.Lock()
-	defer p.receiveMu.Unlock()
-
 	messages, err := p.receiver.ReceiveMessages(ctx, maxMessages, nil)
 	if err != nil {
 		return fmt.Errorf("failed to receive messages: %w", err)
diff --git a/v2/processor_test.go b/v2/processor_test.go
index ab560e47..2aef3cc5 100644
--- a/v2/processor_test.go
+++ b/v2/processor_test.go
@@ -452,19 +452,15 @@ func TestProcessorClose_StopsStartReceiveLoop(t *testing.T) {
 	g.Expect(rcv.ReceiveCalls).To(HaveLen(1))
 }
 
-func TestProcessorClose_WaitsForReceiveBatchBeforeAbandoning(t *testing.T) {
+func TestProcessorClose_DoesNotWaitForActiveReceive(t *testing.T) {
 	messages := make(chan *azservicebus.ReceivedMessage)
-	settler := &fakeSettler{}
 	rcv := &fakeReceiver{
-		fakeSettler:           settler,
+		fakeSettler:           &fakeSettler{},
 		SetupReceivedMessages: messages,
 		SetupMaxReceiveCalls:  10,
 		SetupReceiveStarted:   make(chan struct{}, 1),
 	}
-	releaseHandler := make(chan struct{})
-	processor := shuttle.NewProcessor(rcv, func(ctx context.Context, settler shuttle.MessageSettler, message *azservicebus.ReceivedMessage) {
-		<-releaseHandler
-	}, &shuttle.ProcessorOptions{
+	processor := shuttle.NewProcessor(rcv, MyHandler(0*time.Second), &shuttle.ProcessorOptions{
 		ReceiveInterval: to.Ptr(1 * time.Hour),
 	})
 	errCh := make(chan error, 1)
@@ -475,17 +471,10 @@ func TestProcessorClose_WaitsForReceiveBatchBeforeAbandoning(t *testing.T) {
 
 	closeErrCh := make(chan error, 1)
 	go func() { closeErrCh <- processor.Close(context.Background()) }()
-	g.Consistently(closeErrCh, 20*time.Millisecond).ShouldNot(Receive())
-
-	message := &azservicebus.ReceivedMessage{MessageID: "received-during-close"}
-	messages <- message
 
 	g.Eventually(closeErrCh).Should(Receive(Succeed()))
-	abandoned := settler.abandonedMessages()
-	g.Expect(abandoned).To(HaveLen(1))
-	g.Expect(abandoned[0]).To(BeIdenticalTo(message))
 
-	close(releaseHandler)
+	close(messages)
 	g.Eventually(errCh).Should(Receive(MatchError(context.Canceled)))
 }
 

From 594e4c5efa8447a9aa1ba9ca875122f4f3c9743a Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 13:57:49 -0700
Subject: [PATCH 09/15] Simplify processor close review cleanup

---
 v2/inflightmessages.go | 9 ++-------
 v2/processor.go        | 9 ++-------
 v2/processor_test.go   | 2 --
 3 files changed, 4 insertions(+), 16 deletions(-)

diff --git a/v2/inflightmessages.go b/v2/inflightmessages.go
index 3fc155cc..0cdeaece 100644
--- a/v2/inflightmessages.go
+++ b/v2/inflightmessages.go
@@ -55,19 +55,14 @@ func (m *inFlightMessages) close(ctx context.Context, settler MessageSettler) er
 	}
 
 	var errs []error
-	for completed := 0; completed < len(messages); completed++ {
+	for range messages {
 		select {
 		case err := <-abandonResults:
 			if err != nil {
 				errs = append(errs, err)
 			}
 		case <-ctx.Done():
-			getLogger(ctx).Warn(fmt.Sprintf(
-				"processor close context done while abandoning in-flight messages: %s; completed=%d remaining=%d",
-				ctx.Err(),
-				completed,
-				len(messages)-completed,
-			))
+			getLogger(ctx).Warn(fmt.Sprintf("processor close context done while abandoning in-flight messages: %s", ctx.Err()))
 			return errors.Join(append(errs, ctx.Err())...)
 		}
 	}
diff --git a/v2/processor.go b/v2/processor.go
index 93dfc55a..eef66e33 100644
--- a/v2/processor.go
+++ b/v2/processor.go
@@ -48,11 +48,6 @@ type Processor struct {
 	lifecycle         processorLifecycle
 }
 
-// ErrProcessorClosed is returned when Start is called after the processor has been closed.
-var ErrProcessorClosed = errors.New("processor has already been closed")
-
-var errProcessorClosed = fmt.Errorf("%w: %w", ErrProcessorClosed, context.Canceled)
-
 type processorLifecycle struct {
 	mu             sync.Mutex
 	closed         bool
@@ -166,7 +161,7 @@ func NewProcessor(receiver Receiver, handler HandlerFunc, options *ProcessorOpti
 func (p *Processor) Start(ctx context.Context) (err error) {
 	ctx, cancel := p.lifecycle.start(ctx)
 	if ctx == nil {
-		return errProcessorClosed
+		return fmt.Errorf("processor has already been closed: %w", context.Canceled)
 	}
 	defer p.lifecycle.clearStart()
 	defer cancel()
@@ -226,7 +221,7 @@ func (p *Processor) start(ctx context.Context) error {
 			if ctx.Err() != nil || maxMessages == 0 {
 				break
 			}
-			if err := p.receiveAndProcess(ctx, maxMessages, "from processor loop"); err != nil {
+			if err := p.receiveAndProcess(ctx, maxMessages, "processor loop"); err != nil {
 				return err
 			}
 		case <-ctx.Done():
diff --git a/v2/processor_test.go b/v2/processor_test.go
index 2aef3cc5..9fb24026 100644
--- a/v2/processor_test.go
+++ b/v2/processor_test.go
@@ -113,8 +113,6 @@ func TestProcessorStart_ReturnsProcessorClosedErrorAfterClose(t *testing.T) {
 
 	err := processor.Start(context.Background())
 
-	require.ErrorIs(t, err, shuttle.ErrProcessorClosed)
-	require.ErrorIs(t, err, context.Canceled)
 	require.ErrorContains(t, err, "processor has already been closed")
 	require.Empty(t, rcv.ReceiveCalls)
 }

From fb8133ca317c225ef5f377c1957f8604bbadf8a3 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 14:04:20 -0700
Subject: [PATCH 10/15] Clarify closed processor start error

---
 v2/processor.go      | 2 +-
 v2/processor_test.go | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/v2/processor.go b/v2/processor.go
index eef66e33..8954ed09 100644
--- a/v2/processor.go
+++ b/v2/processor.go
@@ -161,7 +161,7 @@ func NewProcessor(receiver Receiver, handler HandlerFunc, options *ProcessorOpti
 func (p *Processor) Start(ctx context.Context) (err error) {
 	ctx, cancel := p.lifecycle.start(ctx)
 	if ctx == nil {
-		return fmt.Errorf("processor has already been closed: %w", context.Canceled)
+		return fmt.Errorf("failed to start processor: processor has already been closed: %w", context.Canceled)
 	}
 	defer p.lifecycle.clearStart()
 	defer cancel()
diff --git a/v2/processor_test.go b/v2/processor_test.go
index 9fb24026..ae08c75e 100644
--- a/v2/processor_test.go
+++ b/v2/processor_test.go
@@ -113,7 +113,7 @@ func TestProcessorStart_ReturnsProcessorClosedErrorAfterClose(t *testing.T) {
 
 	err := processor.Start(context.Background())
 
-	require.ErrorContains(t, err, "processor has already been closed")
+	require.ErrorContains(t, err, "failed to start processor: processor has already been closed")
 	require.Empty(t, rcv.ReceiveCalls)
 }
 

From fcfb2e2715e20ad660fa7386d0b99a0b428ef0b0 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 14:11:08 -0700
Subject: [PATCH 11/15] Polish closed processor start error

---
 v2/processor.go      | 2 +-
 v2/processor_test.go | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/v2/processor.go b/v2/processor.go
index 8954ed09..2e6bec14 100644
--- a/v2/processor.go
+++ b/v2/processor.go
@@ -161,7 +161,7 @@ func NewProcessor(receiver Receiver, handler HandlerFunc, options *ProcessorOpti
 func (p *Processor) Start(ctx context.Context) (err error) {
 	ctx, cancel := p.lifecycle.start(ctx)
 	if ctx == nil {
-		return fmt.Errorf("failed to start processor: processor has already been closed: %w", context.Canceled)
+		return fmt.Errorf("failed to start processor since it has already been closed: %w", context.Canceled)
 	}
 	defer p.lifecycle.clearStart()
 	defer cancel()
diff --git a/v2/processor_test.go b/v2/processor_test.go
index ae08c75e..f77e5b09 100644
--- a/v2/processor_test.go
+++ b/v2/processor_test.go
@@ -113,7 +113,7 @@ func TestProcessorStart_ReturnsProcessorClosedErrorAfterClose(t *testing.T) {
 
 	err := processor.Start(context.Background())
 
-	require.ErrorContains(t, err, "failed to start processor: processor has already been closed")
+	require.ErrorContains(t, err, "failed to start processor since it has already been closed")
 	require.Empty(t, rcv.ReceiveCalls)
 }
 

From 7513fe564d1301d84284a95840cd25c8fa21c171 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 14:40:29 -0700
Subject: [PATCH 12/15] Simplify in-flight message close tests

---
 v2/inflightmessages_test.go | 200 ++++++++----------------------------
 1 file changed, 44 insertions(+), 156 deletions(-)

diff --git a/v2/inflightmessages_test.go b/v2/inflightmessages_test.go
index 531cc4b0..88da8300 100644
--- a/v2/inflightmessages_test.go
+++ b/v2/inflightmessages_test.go
@@ -21,6 +21,8 @@ func TestInFlightMessages_CloseAbandonsTrackedMessages(t *testing.T) {
 	inFlight.track(forgottenMessage)
 	inFlight.forget(forgottenMessage)
 
+	require.NoError(t, inFlight.close(context.Background(), settler))
+	require.Empty(t, inFlight.messages())
 	require.NoError(t, inFlight.close(context.Background(), settler))
 
 	messages := settler.abandonedMessages()
@@ -35,9 +37,21 @@ func TestInFlightMessages_CloseAbandonsTrackedMessagesConcurrently(t *testing.T)
 	abandonStarted := make(chan *azservicebus.ReceivedMessage, 2)
 	releaseAbandons := make(chan struct{})
 	var releaseOnce sync.Once
-	settler := &blockingInFlightMessageSettler{
-		abandonStarted: abandonStarted,
-		releaseAbandon: releaseAbandons,
+	settler := &inFlightMessageSettler{
+		abandon: func(ctx context.Context, message *azservicebus.ReceivedMessage) error {
+			select {
+			case abandonStarted <- message:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+
+			select {
+			case <-releaseAbandons:
+				return nil
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		},
 	}
 
 	release := func() {
@@ -82,9 +96,12 @@ func TestInFlightMessages_CloseReturnsWhenContextIsCanceled(t *testing.T) {
 	abandonStarted := make(chan *azservicebus.ReceivedMessage, 1)
 	releaseAbandon := make(chan struct{})
 	var releaseOnce sync.Once
-	settler := &contextIgnoringBlockingInFlightMessageSettler{
-		abandonStarted: abandonStarted,
-		releaseAbandon: releaseAbandon,
+	settler := &inFlightMessageSettler{
+		abandon: func(ctx context.Context, message *azservicebus.ReceivedMessage) error {
+			abandonStarted <- message
+			<-releaseAbandon
+			return nil
+		},
 	}
 
 	release := func() {
@@ -123,111 +140,53 @@ func TestInFlightMessages_CloseReturnsWhenContextIsCanceled(t *testing.T) {
 	}, 1*time.Second, 10*time.Millisecond)
 }
 
-func TestInFlightMessages_CloseRemovesTrackedMessages(t *testing.T) {
-	inFlight := newInFlightMessages()
-	settler := &inFlightMessageSettler{}
-	message := &azservicebus.ReceivedMessage{MessageID: "message"}
-
-	inFlight.track(message)
-
-	require.NoError(t, inFlight.close(context.Background(), settler))
-	require.NoError(t, inFlight.close(context.Background(), settler))
-
-	messages := settler.abandonedMessages()
-	require.Len(t, messages, 1)
-	require.Same(t, message, messages[0])
-}
-
-func TestInFlightMessages_CloseReturnsAbandonErrors(t *testing.T) {
+func TestInFlightMessages_CloseReturnsAbandonErrorsAndKeepsMessages(t *testing.T) {
 	firstErr := errors.New("first abandon failed")
 	secondErr := errors.New("second abandon failed")
 	inFlight := newInFlightMessages()
+	firstMessage := &azservicebus.ReceivedMessage{MessageID: "first"}
+	secondMessage := &azservicebus.ReceivedMessage{MessageID: "second"}
+	abandonErrors := make(chan error, 2)
+	abandonErrors <- firstErr
+	abandonErrors <- secondErr
 	settler := &inFlightMessageSettler{
-		abandonErrors: []error{firstErr, secondErr},
+		abandon: func(ctx context.Context, message *azservicebus.ReceivedMessage) error {
+			return <-abandonErrors
+		},
 	}
 
-	inFlight.track(&azservicebus.ReceivedMessage{MessageID: "first"})
-	inFlight.track(&azservicebus.ReceivedMessage{MessageID: "second"})
+	inFlight.track(firstMessage)
+	inFlight.track(secondMessage)
 
 	err := inFlight.close(context.Background(), settler)
 
 	require.Error(t, err)
 	require.ErrorIs(t, err, firstErr)
 	require.ErrorIs(t, err, secondErr)
-}
-
-func TestInFlightMessages_CloseKeepsMessagesWhenAbandonFails(t *testing.T) {
-	abandonErr := errors.New("abandon failed")
-	inFlight := newInFlightMessages()
-	failingSettler := &inFlightMessageSettler{
-		abandonErrors: []error{abandonErr},
-	}
-	message := &azservicebus.ReceivedMessage{MessageID: "message"}
-
-	inFlight.track(message)
-
-	err := inFlight.close(context.Background(), failingSettler)
-
-	require.ErrorIs(t, err, abandonErr)
-	messages := inFlight.messages()
-	require.Len(t, messages, 1)
-	require.Same(t, message, messages[0])
+	require.ElementsMatch(t, []*azservicebus.ReceivedMessage{firstMessage, secondMessage}, inFlight.messages())
 
 	successfulSettler := &inFlightMessageSettler{}
 	require.NoError(t, inFlight.close(context.Background(), successfulSettler))
 	require.Empty(t, inFlight.messages())
 	abandonedMessages := successfulSettler.abandonedMessages()
-	require.Len(t, abandonedMessages, 1)
-	require.Same(t, message, abandonedMessages[0])
-}
-
-func TestInFlightMessages_CloseUsesCloseContext(t *testing.T) {
-	inFlight := newInFlightMessages()
-	settler := &inFlightMessageSettler{}
-
-	inFlight.track(&azservicebus.ReceivedMessage{MessageID: "message"})
-
-	deadline := time.Now().Add(1 * time.Minute)
-	ctx, cancel := context.WithDeadline(context.Background(), deadline)
-	defer cancel()
-
-	require.NoError(t, inFlight.close(ctx, settler))
-
-	deadlines := settler.abandonDeadlines()
-	require.Len(t, deadlines, 1)
-	require.True(t, deadlines[0].ok)
-	require.True(t, deadlines[0].deadline.Equal(deadline))
+	require.ElementsMatch(t, []*azservicebus.ReceivedMessage{firstMessage, secondMessage}, abandonedMessages)
 }
 
 type inFlightMessageSettler struct {
-	mu             sync.Mutex
-	abandoned      []*azservicebus.ReceivedMessage
-	deadlines      []abandonDeadline
-	abandonErrors  []error
-	abandonAttempt int
-}
-
-type abandonDeadline struct {
-	deadline time.Time
-	ok       bool
+	mu        sync.Mutex
+	abandoned []*azservicebus.ReceivedMessage
+	abandon   func(context.Context, *azservicebus.ReceivedMessage) error
 }
 
 func (s *inFlightMessageSettler) AbandonMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.AbandonMessageOptions) error {
-	deadline, ok := ctx.Deadline()
-
 	s.mu.Lock()
-	defer s.mu.Unlock()
 	s.abandoned = append(s.abandoned, message)
-	s.deadlines = append(s.deadlines, abandonDeadline{
-		deadline: deadline,
-		ok:       ok,
-	})
-	err := error(nil)
-	if s.abandonAttempt < len(s.abandonErrors) {
-		err = s.abandonErrors[s.abandonAttempt]
+	s.mu.Unlock()
+
+	if s.abandon != nil {
+		return s.abandon(ctx, message)
 	}
-	s.abandonAttempt++
-	return err
+	return nil
 }
 
 func (s *inFlightMessageSettler) CompleteMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.CompleteMessageOptions) error {
@@ -253,74 +212,3 @@ func (s *inFlightMessageSettler) abandonedMessages() []*azservicebus.ReceivedMes
 	copy(messages, s.abandoned)
 	return messages
 }
-
-func (s *inFlightMessageSettler) abandonDeadlines() []abandonDeadline {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	deadlines := make([]abandonDeadline, len(s.deadlines))
-	copy(deadlines, s.deadlines)
-	return deadlines
-}
-
-type blockingInFlightMessageSettler struct {
-	abandonStarted chan<- *azservicebus.ReceivedMessage
-	releaseAbandon <-chan struct{}
-}
-
-func (s *blockingInFlightMessageSettler) AbandonMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.AbandonMessageOptions) error {
-	select {
-	case s.abandonStarted <- message:
-	case <-ctx.Done():
-		return ctx.Err()
-	}
-
-	select {
-	case <-s.releaseAbandon:
-		return nil
-	case <-ctx.Done():
-		return ctx.Err()
-	}
-}
-
-func (s *blockingInFlightMessageSettler) CompleteMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.CompleteMessageOptions) error {
-	return nil
-}
-
-func (s *blockingInFlightMessageSettler) DeadLetterMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.DeadLetterOptions) error {
-	return nil
-}
-
-func (s *blockingInFlightMessageSettler) DeferMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.DeferMessageOptions) error {
-	return nil
-}
-
-func (s *blockingInFlightMessageSettler) RenewMessageLock(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.RenewMessageLockOptions) error {
-	return nil
-}
-
-type contextIgnoringBlockingInFlightMessageSettler struct {
-	abandonStarted chan<- *azservicebus.ReceivedMessage
-	releaseAbandon <-chan struct{}
-}
-
-func (s *contextIgnoringBlockingInFlightMessageSettler) AbandonMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.AbandonMessageOptions) error {
-	s.abandonStarted <- message
-	<-s.releaseAbandon
-	return nil
-}
-
-func (s *contextIgnoringBlockingInFlightMessageSettler) CompleteMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.CompleteMessageOptions) error {
-	return nil
-}
-
-func (s *contextIgnoringBlockingInFlightMessageSettler) DeadLetterMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.DeadLetterOptions) error {
-	return nil
-}
-
-func (s *contextIgnoringBlockingInFlightMessageSettler) DeferMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.DeferMessageOptions) error {
-	return nil
-}
-
-func (s *contextIgnoringBlockingInFlightMessageSettler) RenewMessageLock(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.RenewMessageLockOptions) error {
-	return nil
-}

From 46f603acab24406968f596731bb51caa62ad60c2 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 15:01:15 -0700
Subject: [PATCH 13/15] Cover in-flight close context deadline

---
 v2/inflightmessages_test.go | 72 ++++++++++++++++++++++++++++++-------
 1 file changed, 60 insertions(+), 12 deletions(-)

diff --git a/v2/inflightmessages_test.go b/v2/inflightmessages_test.go
index 88da8300..311d29ee 100644
--- a/v2/inflightmessages_test.go
+++ b/v2/inflightmessages_test.go
@@ -39,18 +39,9 @@ func TestInFlightMessages_CloseAbandonsTrackedMessagesConcurrently(t *testing.T)
 	var releaseOnce sync.Once
 	settler := &inFlightMessageSettler{
 		abandon: func(ctx context.Context, message *azservicebus.ReceivedMessage) error {
-			select {
-			case abandonStarted <- message:
-			case <-ctx.Done():
-				return ctx.Err()
-			}
-
-			select {
-			case <-releaseAbandons:
-				return nil
-			case <-ctx.Done():
-				return ctx.Err()
-			}
+			abandonStarted <- message
+			<-releaseAbandons
+			return nil
 		},
 	}
 
@@ -134,6 +125,63 @@ func TestInFlightMessages_CloseReturnsWhenContextIsCanceled(t *testing.T) {
 		t.Fatal("timed out waiting for close to return after context cancellation")
 	}
 
+	abandonedMessages := settler.abandonedMessages()
+	require.Len(t, abandonedMessages, 1)
+	require.Equal(t, "blocked", abandonedMessages[0].MessageID)
+
+	release()
+	require.Eventually(t, func() bool {
+		return len(inFlight.messages()) == 0
+	}, 1*time.Second, 10*time.Millisecond)
+}
+
+func TestInFlightMessages_CloseReturnsWhenContextDeadlineExpires(t *testing.T) {
+	inFlight := newInFlightMessages()
+	abandonStarted := make(chan *azservicebus.ReceivedMessage, 1)
+	releaseAbandon := make(chan struct{})
+	var releaseOnce sync.Once
+	settler := &inFlightMessageSettler{
+		abandon: func(ctx context.Context, message *azservicebus.ReceivedMessage) error {
+			abandonStarted <- message
+			<-releaseAbandon
+			return nil
+		},
+	}
+
+	release := func() {
+		releaseOnce.Do(func() {
+			close(releaseAbandon)
+		})
+	}
+	t.Cleanup(release)
+
+	inFlight.track(&azservicebus.ReceivedMessage{MessageID: "deadline"})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+	defer cancel()
+
+	closeErr := make(chan error, 1)
+	go func() {
+		closeErr <- inFlight.close(ctx, settler)
+	}()
+
+	select {
+	case <-abandonStarted:
+	case <-time.After(1 * time.Second):
+		t.Fatal("timed out waiting for abandon attempt to start")
+	}
+
+	select {
+	case err := <-closeErr:
+		require.ErrorIs(t, err, context.DeadlineExceeded)
+	case <-time.After(1 * time.Second):
+		t.Fatal("timed out waiting for close to return after context deadline")
+	}
+
+	abandonedMessages := settler.abandonedMessages()
+	require.Len(t, abandonedMessages, 1)
+	require.Equal(t, "deadline", abandonedMessages[0].MessageID)
+
 	release()
 	require.Eventually(t, func() bool {
 		return len(inFlight.messages()) == 0

From e6ca2d8077e90c1a6b32e939f6f85c5acb979437 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 17:31:25 -0700
Subject: [PATCH 14/15] Make fake processor receive respect context

---
 v2/processor.go           |  5 +++--
 v2/processor_fake_test.go | 27 ++++++++-------------------
 v2/processor_test.go      | 26 ++++++--------------------
 3 files changed, 17 insertions(+), 41 deletions(-)

diff --git a/v2/processor.go b/v2/processor.go
index 2e6bec14..79aa149e 100644
--- a/v2/processor.go
+++ b/v2/processor.go
@@ -83,7 +83,7 @@ func (l *processorLifecycle) markClosed() context.CancelFunc {
 	return cancelStartCtx
 }
 
-func (l *processorLifecycle) clearStart() {
+func (l *processorLifecycle) clearCancelStartCtx() {
 	l.mu.Lock()
 	defer l.mu.Unlock()
 
@@ -163,7 +163,7 @@ func (p *Processor) Start(ctx context.Context) (err error) {
 	if ctx == nil {
 		return fmt.Errorf("failed to start processor since it has already been closed: %w", context.Canceled)
 	}
-	defer p.lifecycle.clearStart()
+	defer p.lifecycle.clearCancelStartCtx()
 	defer cancel()
 	defer func() {
 		if rec := recover(); rec != nil {
@@ -176,6 +176,7 @@ func (p *Processor) Start(ctx context.Context) (err error) {
 // Close stops receiving new messages, cancels in-flight message handlers, and
 // best-effort abandons messages currently held by the processor.
 func (p *Processor) Close(ctx context.Context) error {
+	// Close the active processor loop so no further messages are received.
 	p.lifecycle.close()
 
 	return p.inFlightMessages.close(ctx, p.receiver)
diff --git a/v2/processor_fake_test.go b/v2/processor_fake_test.go
index 49e53eca..9796b3ac 100644
--- a/v2/processor_fake_test.go
+++ b/v2/processor_fake_test.go
@@ -71,7 +71,6 @@ type fakeReceiver struct {
 	*fakeSettler
 	SetupMaxReceiveCalls int
 	SetupReceivePanic    string
-	SetupRespectContext  bool
 	SetupReceiveStarted  chan struct{}
 }
 
@@ -88,28 +87,18 @@ func (f *fakeReceiver) ReceiveMessages(ctx context.Context, maxMessages int, _ *
 	}
 	var result []*azservicebus.ReceivedMessage
 	for len(result) < maxMessages {
-		if f.SetupRespectContext {
-			select {
-			case msg, ok := <-f.SetupReceivedMessages:
-				if !ok {
-					return f.receiveResult(result)
-				}
-				result = append(result, msg)
-				if len(f.SetupReceivedMessages) == 0 {
-					return f.receiveResult(result)
-				}
-			case <-ctx.Done():
-				return result, ctx.Err()
+		select {
+		case msg, ok := <-f.SetupReceivedMessages:
+			if !ok {
+				return f.receiveResult(result)
 			}
-			continue
-		}
-		for msg := range f.SetupReceivedMessages {
 			result = append(result, msg)
-			if len(result) == maxMessages || len(f.SetupReceivedMessages) == 0 {
-				break
+			if len(f.SetupReceivedMessages) == 0 {
+				return f.receiveResult(result)
 			}
+		case <-ctx.Done():
+			return result, ctx.Err()
 		}
-		break
 	}
 
 	return f.receiveResult(result)
diff --git a/v2/processor_test.go b/v2/processor_test.go
index f77e5b09..577b4411 100644
--- a/v2/processor_test.go
+++ b/v2/processor_test.go
@@ -70,9 +70,7 @@ func TestProcessorStart_DefaultsToMaxConcurrency(t *testing.T) {
 		SetupReceivedMessages: messages,
 	}
 	processor := shuttle.NewProcessor(rcv, MyHandler(0*time.Second), nil)
-	ctx, cancel := context.WithCancel(context.Background())
-	cancel()
-	err := processor.Start(ctx)
+	err := processor.Start(context.Background())
 	a.ErrorContains(err, "max receive calls exceeded")
 	a.Equal(1, len(rcv.ReceiveCalls), "there should be 1 entry in the ReceiveCalls array")
 	a.Equal(1, rcv.ReceiveCalls[0], "the processor should have used the default max concurrency of 1")
@@ -127,10 +125,7 @@ func TestProcessorStart_CanSetMaxConcurrency(t *testing.T) {
 	processor := shuttle.NewProcessor(rcv, MyHandler(0*time.Second), &shuttle.ProcessorOptions{
 		MaxConcurrency: 10,
 	})
-	ctx, cancel := context.WithCancel(context.Background())
-	// pre-cancel the context
-	cancel()
-	err := processor.Start(ctx)
+	err := processor.Start(context.Background())
 	a.ErrorContains(err, "max receive calls exceeded")
 	a.Equal(1, len(rcv.ReceiveCalls), "there should be 1 entry in the ReceiveCalls array")
 	a.Equal(10, rcv.ReceiveCalls[0], "the processor should have used max concurrency of 10")
@@ -191,9 +186,7 @@ func TestProcessorStart_MaxReceiveCountGreaterThanMaxConcurrency(t *testing.T) {
 		MaxConcurrency:  5,
 		MaxReceiveCount: 10,
 	})
-	ctx, cancel := context.WithCancel(context.Background())
-	cancel()
-	err := processor.Start(ctx)
+	err := processor.Start(context.Background())
 	a.ErrorContains(err, "max receive calls exceeded")
 	a.Equal(1, len(rcv.ReceiveCalls), "there should be 1 entry in the ReceiveCalls array")
 	a.Equal(5, rcv.ReceiveCalls[0], "the processor should have used max concurrency as max receive count")
@@ -210,9 +203,7 @@ func TestProcessorStart_DisableMaxReceiveCount(t *testing.T) {
 		MaxConcurrency:  5,
 		MaxReceiveCount: -1,
 	})
-	ctx, cancel := context.WithCancel(context.Background())
-	cancel()
-	err := processor.Start(ctx)
+	err := processor.Start(context.Background())
 	a.ErrorContains(err, "max receive calls exceeded")
 	a.Equal(1, len(rcv.ReceiveCalls), "there should be 1 entry in the ReceiveCalls array")
 	a.Equal(5, rcv.ReceiveCalls[0], "the processor should have used max concurrency as max receive count")
@@ -311,9 +302,7 @@ func TestProcessorStart_DefaultsToStartMaxAttempt(t *testing.T) {
 		SetupReceiveError:     fmt.Errorf("fake receive error"),
 	}
 	processor := shuttle.NewProcessor(rcv, MyHandler(0*time.Second), nil)
-	ctx, cancel := context.WithCancel(context.Background())
-	cancel()
-	err := processor.Start(ctx)
+	err := processor.Start(context.Background())
 	a.ErrorContains(err, "fake receive error")
 	a.Equal(1, len(rcv.ReceiveCalls), "there should be 1 entry in the ReceiveCalls array")
 	a.Equal(1, rcv.ReceiveCalls[0], "the processor should have used the default max concurrency of 1")
@@ -434,7 +423,6 @@ func TestProcessorClose_StopsStartReceiveLoop(t *testing.T) {
 		fakeSettler:           &fakeSettler{},
 		SetupReceivedMessages: make(chan *azservicebus.ReceivedMessage),
 		SetupMaxReceiveCalls:  10,
-		SetupRespectContext:   true,
 		SetupReceiveStarted:   make(chan struct{}, 1),
 	}
 	processor := shuttle.NewProcessor(rcv, MyHandler(0*time.Second), &shuttle.ProcessorOptions{
@@ -471,9 +459,7 @@ func TestProcessorClose_DoesNotWaitForActiveReceive(t *testing.T) {
 	go func() { closeErrCh <- processor.Close(context.Background()) }()
 
 	g.Eventually(closeErrCh).Should(Receive(Succeed()))
-
-	close(messages)
-	g.Eventually(errCh).Should(Receive(MatchError(context.Canceled)))
+	g.Eventually(errCh).Should(Receive(MatchError(MatchRegexp("failed to receive messages: context canceled"))))
 }
 
 func TestProcessorClose_ReturnsAbandonErrors(t *testing.T) {

From bb879d80b16a3a69efc0f2aa05ab8fe0345d0ce4 Mon Sep 17 00:00:00 2001
From: Karen Chen <karenchen@microsoft.com>
Date: Mon, 29 Jun 2026 18:09:46 -0700
Subject: [PATCH 15/15] Simplify processor close tests

---
 v2/processor_fake_test.go | 15 -------------
 v2/processor_test.go      | 47 ++++++++++++++++++---------------------
 2 files changed, 22 insertions(+), 40 deletions(-)

diff --git a/v2/processor_fake_test.go b/v2/processor_fake_test.go
index 9796b3ac..c0d340b8 100644
--- a/v2/processor_fake_test.go
+++ b/v2/processor_fake_test.go
@@ -3,7 +3,6 @@ package shuttle_test
 import (
 	"context"
 	"fmt"
-	"sync"
 	"sync/atomic"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
@@ -20,16 +19,10 @@ type fakeSettler struct {
 	DeferCalled      atomic.Int32
 	RenewCalled      atomic.Int32
 	SetupAbandonErr  error
-
-	mu                sync.Mutex
-	AbandonedMessages []*azservicebus.ReceivedMessage
 }
 
 func (f *fakeSettler) AbandonMessage(ctx context.Context, message *azservicebus.ReceivedMessage, options *azservicebus.AbandonMessageOptions) error {
 	f.AbandonCalled.Add(1)
-	f.mu.Lock()
-	f.AbandonedMessages = append(f.AbandonedMessages, message)
-	f.mu.Unlock()
 	return f.SetupAbandonErr
 }
 
@@ -53,14 +46,6 @@ func (f *fakeSettler) RenewMessageLock(ctx context.Context, message *azservicebu
 	return nil
 }
 
-func (f *fakeSettler) abandonedMessages() []*azservicebus.ReceivedMessage {
-	f.mu.Lock()
-	defer f.mu.Unlock()
-	messages := make([]*azservicebus.ReceivedMessage, len(f.AbandonedMessages))
-	copy(messages, f.AbandonedMessages)
-	return messages
-}
-
 type fakeReceiver struct {
 	// outcomes to verify
 	ReceiveCalls []int // array of maxMessage value passed to receive calls in the lifetime of the fake receiver
diff --git a/v2/processor_test.go b/v2/processor_test.go
index 577b4411..7ae2932d 100644
--- a/v2/processor_test.go
+++ b/v2/processor_test.go
@@ -366,6 +366,28 @@ func TestProcessorStart_ContextCanceledDuringStartRetry(t *testing.T) {
 	a.Equal(1, rcv.ReceiveCalls[1], "the processor should have retried the receive call once")
 }
 
+func TestProcessorStart_DoesNotRetryAfterCloseDuringFirstAttempt(t *testing.T) {
+	rcv := &fakeReceiver{
+		fakeSettler:           &fakeSettler{},
+		SetupReceivedMessages: make(chan *azservicebus.ReceivedMessage),
+		SetupMaxReceiveCalls:  10,
+		SetupReceiveStarted:   make(chan struct{}, 1),
+	}
+
+	processor := shuttle.NewProcessor(rcv, MyHandler(0*time.Second), &shuttle.ProcessorOptions{
+		StartMaxAttempt:         3,
+		StartRetryDelayStrategy: &shuttle.ConstantDelayStrategy{Delay: 20 * time.Millisecond},
+	})
+	errCh := make(chan error, 1)
+	go func() { errCh <- processor.Start(context.Background()) }()
+
+	g := NewWithT(t)
+	g.Eventually(rcv.SetupReceiveStarted).Should(Receive())
+	g.Expect(processor.Close(context.Background())).To(Succeed())
+	g.Eventually(errCh).Should(Receive(MatchError(context.Canceled)))
+	g.Expect(rcv.ReceiveCalls).To(HaveLen(1))
+}
+
 func TestProcessorStart_RecoversReceiverPanic(t *testing.T) {
 	rcv := &fakeReceiver{
 		fakeSettler:           &fakeSettler{},
@@ -415,7 +437,6 @@ func TestProcessorClose_CancelsAndAbandonsInflightMessages(t *testing.T) {
 	g.Eventually(canceled).Should(Receive())
 	g.Eventually(errCh).Should(Receive(MatchError(context.Canceled)))
 	g.Expect(settler.AbandonCalled.Load()).To(Equal(int32(2)))
-	g.Expect(settler.abandonedMessages()).To(HaveLen(2))
 }
 
 func TestProcessorClose_StopsStartReceiveLoop(t *testing.T) {
@@ -438,30 +459,6 @@ func TestProcessorClose_StopsStartReceiveLoop(t *testing.T) {
 	g.Expect(rcv.ReceiveCalls).To(HaveLen(1))
 }
 
-func TestProcessorClose_DoesNotWaitForActiveReceive(t *testing.T) {
-	messages := make(chan *azservicebus.ReceivedMessage)
-	rcv := &fakeReceiver{
-		fakeSettler:           &fakeSettler{},
-		SetupReceivedMessages: messages,
-		SetupMaxReceiveCalls:  10,
-		SetupReceiveStarted:   make(chan struct{}, 1),
-	}
-	processor := shuttle.NewProcessor(rcv, MyHandler(0*time.Second), &shuttle.ProcessorOptions{
-		ReceiveInterval: to.Ptr(1 * time.Hour),
-	})
-	errCh := make(chan error, 1)
-	go func() { errCh <- processor.Start(context.Background()) }()
-
-	g := NewWithT(t)
-	g.Eventually(rcv.SetupReceiveStarted).Should(Receive())
-
-	closeErrCh := make(chan error, 1)
-	go func() { closeErrCh <- processor.Close(context.Background()) }()
-
-	g.Eventually(closeErrCh).Should(Receive(Succeed()))
-	g.Eventually(errCh).Should(Receive(MatchError(MatchRegexp("failed to receive messages: context canceled"))))
-}
-
 func TestProcessorClose_ReturnsAbandonErrors(t *testing.T) {
 	abandonErr := errors.New("abandon failed")
 	messages := messagesChannel(2)