Retry ForwardPoll on ResourceExhausted instead of disabling forwarding (#10019)

## What

On child partitions, when `ForwardPoll` gets a `ResourceExhausted`
error, re-enqueue the poller with forwarding still enabled so it
retries.

## Why

When `frontend.enableCancelWorkerPollsOnShutdown` is enabled, a wave of
cancelled polls followed by re-polls can trigger rate limiting on the
root partition. The `forwardPolls` goroutine was treating
`ResourceExhausted` the same as other errors — permanently disabling
forwarding by setting `forwardCtx = nil`. This caused the poll to fall
back to waiting for a local task match, which on a child partition with
no backlog means waiting the full 60s long-poll timeout before the
poller retries.

The `forwardTasks` goroutine already had proper `ResourceExhausted`
retry logic with exponential backoff; `forwardPolls` was missing it.

## How did you test it?

Unit test (`TestForwardPollRetriesOnResourceExhausted`) that sets up a
child partition with a mock matching client returning
`ResourceExhausted` on the first `ForwardPoll` call and a valid task on
the second, then verifies the poll succeeds via retry.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: rkannan82

common/dynamicconfig/constants.go

@@ -1450,6 +1450,11 @@ second per poller by one physical queue manager`,
 		60*time.Second,
 		`Timeout for forwarded backlog task (requires new matcher)`,
 	)
+	MatchingForwardPollRetryMaxInterval = NewTaskQueueDurationSetting(
+		"matching.forwardPollRetryMaxInterval",
+		10*time.Second,
+		`Max backoff interval when retrying a rate-limited ForwardPoll from a child partition`,
+	)
 	MatchingFairnessCounter = NewTaskQueueTypedSetting(
 		"matching.fairnessCounter",
 		counter.DefaultCounterParams,

service/matching/config.go

@@ -97,11 +97,12 @@ type (
 		MaxFairnessKeyWeightOverrides dynamicconfig.IntPropertyFnWithTaskQueueFilter
 
 		// Time to hold a poll request before returning an empty response if there are no tasks
-		LongPollExpirationInterval dynamicconfig.DurationPropertyFnWithTaskQueueFilter
-		BacklogTaskForwardTimeout  dynamicconfig.DurationPropertyFnWithTaskQueueFilter
-		MinTaskThrottlingBurstSize dynamicconfig.IntPropertyFnWithTaskQueueFilter
-		MaxTaskDeleteBatchSize     dynamicconfig.IntPropertyFnWithTaskQueueFilter
-		TaskDeleteInterval         dynamicconfig.DurationPropertyFnWithTaskQueueFilter
+		LongPollExpirationInterval  dynamicconfig.DurationPropertyFnWithTaskQueueFilter
+		BacklogTaskForwardTimeout   dynamicconfig.DurationPropertyFnWithTaskQueueFilter
+		ForwardPollRetryMaxInterval dynamicconfig.DurationPropertyFnWithTaskQueueFilter
+		MinTaskThrottlingBurstSize  dynamicconfig.IntPropertyFnWithTaskQueueFilter
+		MaxTaskDeleteBatchSize      dynamicconfig.IntPropertyFnWithTaskQueueFilter
+		TaskDeleteInterval          dynamicconfig.DurationPropertyFnWithTaskQueueFilter
 
 		// taskWriter configuration
 		OutstandingTaskAppendsThreshold dynamicconfig.IntPropertyFnWithTaskQueueFilter
@@ -156,6 +157,7 @@ type (
 		// Time to hold a poll request before returning an empty response if there are no tasks
 		LongPollExpirationInterval     func() time.Duration
 		BacklogTaskForwardTimeout      func() time.Duration
+		ForwardPollRetryMaxInterval    func() time.Duration
 		RangeSize                      int64
 		NewMatcher                     bool
 		NewMatcherSub                  func(func(dynamicconfig.GradualChange[bool])) (dynamicconfig.GradualChange[bool], func())
@@ -285,6 +287,7 @@ func NewConfig(
 		MaxTaskQueueIdleTime:                     dynamicconfig.MatchingMaxTaskQueueIdleTime.Get(dc),
 		LongPollExpirationInterval:               dynamicconfig.MatchingLongPollExpirationInterval.Get(dc),
 		BacklogTaskForwardTimeout:                dynamicconfig.MatchingBacklogTaskForwardTimeout.Get(dc),
+		ForwardPollRetryMaxInterval:              dynamicconfig.MatchingForwardPollRetryMaxInterval.Get(dc),
 		MinTaskThrottlingBurstSize:               dynamicconfig.MatchingMinTaskThrottlingBurstSize.Get(dc),
 		MaxTaskDeleteBatchSize:                   dynamicconfig.MatchingMaxTaskDeleteBatchSize.Get(dc),
 		TaskDeleteInterval:                       dynamicconfig.MatchingTaskDeleteInterval.Get(dc),
@@ -439,6 +442,9 @@ func newTaskQueueConfig(tq *tqid.TaskQueue, config *Config, ns namespace.Name) *
 		BacklogTaskForwardTimeout: func() time.Duration {
 			return config.BacklogTaskForwardTimeout(ns.String(), taskQueueName, taskType)
 		},
+		ForwardPollRetryMaxInterval: func() time.Duration {
+			return config.ForwardPollRetryMaxInterval(ns.String(), taskQueueName, taskType)
+		},
 		MaxTaskDeleteBatchSize: func() int {
 			return config.MaxTaskDeleteBatchSize(ns.String(), taskQueueName, taskType)
 		},

service/matching/pri_matcher.go

@@ -293,6 +293,10 @@ func (tm *priTaskMatcher) forwardPolls(
 	ft pollForwarderType,
 	target *tqid.NormalPartition,
 ) {
+	policy := backoff.NewExponentialRetryPolicy(time.Second).
+		WithMaximumInterval(tm.config.ForwardPollRetryMaxInterval()).
+		WithExpirationInterval(backoff.NoInterval)
+	retrier := backoff.NewRetrier(policy, clock.NewRealTimeSource())
 	forwarderTask := newPollForwarderTask(effectivePriority, ft)
 	ctxs := []context.Context{ctx} // ctx should be equal to or child of tm.tqCtx
 	for ctx.Err() == nil {
@@ -332,6 +336,7 @@ func (tm *priTaskMatcher) forwardPolls(
 		_ = stop()
 		if err == nil {
 			tm.data.FinishMatchAfterPollForward(poller, task)
+			retrier.Reset()
 			if ft == priorityBacklogPollForwarder {
 				metrics.PriorityBacklogForwardedPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
 			}
@@ -343,6 +348,10 @@ func (tm *priTaskMatcher) forwardPolls(
 			// 4× to allow for a few rounds plus propagation.
 			interval := cmp.Or(tm.config.EphemeralDataUpdateInterval(), time.Minute)
 			_ = util.InterruptibleSleep(ctx, 4*interval)
+		} else if common.IsResourceExhausted(err) {
+			// Rate limited: re-enqueue with forwarding still enabled so it retries.
+			tm.data.ReenqueuePollerIfNotMatched(poller)
+			_ = util.InterruptibleSleep(ctx, retrier.NextBackOff(err))
 		} else {
 			// Re-enqueue to let it match again, if it hasn't gotten a context timeout already.
 			poller.forwardCtx = nil // disable forwarding next time

service/matching/pri_matcher_test.go

@@ -4,14 +4,20 @@ import (
 	"context"
 	"sync/atomic"
 	"testing"
+	"testing/synctest"
 	"time"
 
+	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/api/matchingservice/v1"
+	"go.temporal.io/server/api/matchingservicemock/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
+	"go.temporal.io/server/common/testing/testhooks"
 	"go.temporal.io/server/common/testing/testlogger"
 	"go.temporal.io/server/common/tqid"
 	"go.uber.org/mock/gomock"
@@ -103,3 +109,86 @@ func (s *PriMatcherSuite) TestValidatorWorksOnRoot() {
 
 	s.True(validatorValidatedTask.Load(), "Validator should have called maybeValidate")
 }
+
+// TestForwardPollRetriesOnResourceExhausted verifies that when a child partition's
+// ForwardPoll gets a ResourceExhausted error (rate limited), the poller is re-enqueued
+// with forwarding still enabled and retries until it succeeds. This is a regression test
+// for a bug where ForwardPoll permanently disabled forwarding on transient rate-limit
+// errors, causing polls to wait for the full 60s timeout instead of retrying.
+func (s *PriMatcherSuite) TestForwardPollRetriesOnResourceExhausted() {
+	// Use synctest to virtualize time so the backoff sleep is instant.
+	synctest.Test(s.T(), func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		tq := tqid.UnsafeTaskQueueFamily("nsid", "tq").TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW)
+		childPartition := tq.NormalPartition(1) // child partition /1
+
+		cfg := newTaskQueueConfig(tq, NewConfig(dynamicconfig.NewNoopCollection()), "nsname")
+		// Use a generous poll timeout so we can distinguish retry success from timeout.
+		cfg.LongPollExpirationInterval = func() time.Duration { return 10 * time.Second }
+
+		mockClient := matchingservicemock.NewMockMatchingServiceClient(s.controller)
+
+		// First ForwardPoll call: return ResourceExhausted (simulating rate limit storm).
+		// Second call: return a valid task response.
+		rateLimitErr := serviceerror.NewResourceExhausted(
+			enumspb.RESOURCE_EXHAUSTED_CAUSE_RPS_LIMIT, "rate limit exceeded",
+		)
+		taskToken := []byte("test-task-token")
+
+		gomock.InOrder(
+			mockClient.EXPECT().
+				PollWorkflowTaskQueue(gomock.Any(), gomock.Any(), gomock.Any()).
+				Return(nil, rateLimitErr),
+			mockClient.EXPECT().
+				PollWorkflowTaskQueue(gomock.Any(), gomock.Any(), gomock.Any()).
+				Return(&matchingservice.PollWorkflowTaskQueueResponse{
+					TaskToken: taskToken,
+				}, nil),
+		)
+
+		// Create a priForwarder for the child partition (non-nil fwdr triggers child behavior).
+		queue := UnversionedQueueKey(childPartition)
+		fwdr, err := newPriForwarder(
+			&cfg.forwarderConfig,
+			queue,
+			mockClient,
+			testhooks.TestHooks{},
+		)
+		require.NoError(t, err)
+
+		rateLimitManager := newRateLimitManager(&mockUserDataManager{}, cfg, enumspb.TASK_QUEUE_TYPE_WORKFLOW)
+
+		tm := newPriTaskMatcher(
+			ctx,
+			cfg,
+			childPartition,
+			fwdr,
+			mockClient,
+			nil, // no validator needed on child
+			s.logger,
+			metrics.NoopMetricsHandler,
+			rateLimitManager,
+			func() {},
+		)
+
+		tm.Start()
+		defer tm.Stop()
+
+		// Poll from the child partition. The forwardPolls goroutine should:
+		// 1. Pick up this poller
+		// 2. Try ForwardPoll → get ResourceExhausted
+		// 3. Re-enqueue poller with forwarding still enabled
+		// 4. Try ForwardPoll again → succeed with task token
+		// 5. Return the task to the poller
+		pollCtx, pollCancel := context.WithTimeout(ctx, 5*time.Second)
+		defer pollCancel()
+
+		task, err := tm.Poll(pollCtx, &pollMetadata{})
+		require.NoError(t, err)
+		require.NotNil(t, task, "poll should have received a task via forwarding retry")
+		require.True(t, task.isStarted(), "task should be a started (forwarded) task")
+		require.Equal(t, taskToken, task.started.workflowTaskInfo.TaskToken)
+	})
+}