fix(messagequeue): clarify canceled worker polls

platform/extension/messagequeue/mysql/BUILD.bazel

@@ -52,6 +52,8 @@ go_test(
         "@com_github_stretchr_testify//require:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_mock//gomock:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
         "@org_uber_go_zap//zaptest:go_default_library",
+        "@org_uber_go_zap//zaptest/observer:go_default_library",
     ],
 )

platform/extension/messagequeue/mysql/subscriber.go

@@ -16,6 +16,7 @@ package mysql
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sort"
 	"strconv"
@@ -1014,6 +1015,18 @@ func (w *partitionWorker) run(ctx context.Context) {
 			// return to — the only recovery is to retry on the next tick, which
 			// happens automatically. All pollAndDeliver operations are idempotent.
 			if err := w.pollAndDeliver(ctx); err != nil {
+				// A stopped worker has no next tick to recover on. Its context is
+				// deliberately cancelled to interrupt any in-flight store call, so
+				// the resulting error is part of normal teardown.
+				if errors.Is(err, context.Canceled) && errors.Is(ctx.Err(), context.Canceled) {
+					w.subscriber.logger.Infow("poll canceled while stopping partition worker",
+						"topic", w.sub.topic,
+						"partition_key", w.partitionKey,
+						"consumer_group", w.sub.config.ConsumerGroup,
+						"subscriber_name", w.sub.config.SubscriberName,
+					)
+					return
+				}
 				w.subscriber.logger.Errorw("poll failed",
 					"topic", w.sub.topic,
 					"partition_key", w.partitionKey,

platform/extension/messagequeue/mysql/subscriber_test.go

@@ -25,7 +25,9 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
 	"go.uber.org/mock/gomock"
+	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
+	"go.uber.org/zap/zaptest/observer"
 
 	"github.com/uber/submitqueue/platform/base/failure"
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
@@ -1013,6 +1015,111 @@ func TestSubscriber_StopAllWorkers(t *testing.T) {
 	}
 }
 
+func TestPartitionWorker_RunPollErrorLogging(t *testing.T) {
+	tests := []struct {
+		name             string
+		pollError        func(context.Context) error
+		cancelDuringPoll bool
+		expectedLogError string
+	}{
+		{
+			name: "worker cancellation is informational",
+			pollError: func(ctx context.Context) error {
+				<-ctx.Done()
+				return ctx.Err()
+			},
+			cancelDuringPoll: true,
+		},
+		{
+			name: "store cancellation on active worker is logged",
+			pollError: func(context.Context) error {
+				return context.Canceled
+			},
+			expectedLogError: "get acked offset: context canceled",
+		},
+		{
+			name: "store failure is logged",
+			pollError: func(context.Context) error {
+				return errors.New("store unavailable")
+			},
+			expectedLogError: "get acked offset: store unavailable",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			mockMessageStore := NewMockmessageStore(ctrl)
+			mockOffsetStore := NewMockoffsetStore(ctrl)
+			mockLeaseStore := NewMockpartitionLeaseStore(ctrl)
+
+			pollStarted := make(chan struct{}, 1)
+			mockOffsetStore.EXPECT().Initialize(gomock.Any(), "test_topic", "part-1", "test-consumer").Return(nil)
+			mockOffsetStore.EXPECT().GetAckedOffset(gomock.Any(), "test_topic", "part-1", "test-consumer").DoAndReturn(
+				func(ctx context.Context, _, _, _ string) (int64, error) {
+					select {
+					case pollStarted <- struct{}{}:
+					default:
+					}
+					return 0, tt.pollError(ctx)
+				},
+			).AnyTimes()
+
+			core, logs := observer.New(zap.InfoLevel)
+			s := NewSubscriber(
+				zap.New(core).Sugar(),
+				tally.NoopScope,
+				mockMessageStore,
+				mockOffsetStore,
+				mockLeaseStore,
+				newTestHeartbeatStore(ctrl),
+				newTestDeliveryStateStore(ctrl),
+			)
+			s.OnSignal = make(chan HookSignal, 1)
+			cfg := testSubscriptionConfig()
+			cfg.PollIntervalMs = 1
+			sub := &subscription{
+				topic:      "test_topic",
+				config:     cfg,
+				deliveryCh: make(chan extqueue.Delivery, 1),
+			}
+			worker := &partitionWorker{
+				partitionKey: "part-1",
+				sub:          sub,
+				subscriber:   s,
+				done:         make(chan struct{}),
+			}
+
+			ctx, cancel := context.WithCancel(context.Background())
+			sub.workerWg.Add(1)
+			go worker.run(ctx)
+
+			<-pollStarted
+			if tt.cancelDuringPoll {
+				cancel()
+			} else {
+				<-s.OnSignal
+				cancel()
+			}
+			<-worker.done
+
+			pollFailures := logs.FilterMessage("poll failed").All()
+			stoppedPolls := logs.FilterMessage("poll canceled while stopping partition worker").All()
+			if tt.expectedLogError != "" {
+				require.NotEmpty(t, pollFailures)
+				for _, entry := range pollFailures {
+					assert.Equal(t, zap.ErrorLevel, entry.Level)
+					assert.Equal(t, tt.expectedLogError, entry.ContextMap()["error"])
+				}
+			} else {
+				assert.Empty(t, pollFailures)
+				require.Len(t, stoppedPolls, 1)
+				assert.Equal(t, zap.InfoLevel, stoppedPolls[0].Level)
+			}
+		})
+	}
+}
+
 func TestSubscriber_FairShareCap(t *testing.T) {
 	tests := []struct {
 		name       string