Back off on DescribeWorkerDeployment ResourceExhausted error (#291)

<!--- Note to EXTERNAL Contributors -->
<!-- Thanks for opening a PR! 
If it is a significant code change, please **make sure there is an open
issue** for this.
We work best with you when we have accepted the idea first before you
code. -->

<!--- For ALL Contributors 👇 -->

## What was changed and why

#### Problem

Namespaces with many TemporalWorkerDeployment objects can trigger
Temporal's per-namespace DescribeWorkerDeployment rate limit
(frontend.globalNamespaceWorkerDeploymentReadRPS, default 50 RPS). When
this happens, the reconciler was returning the error immediately,
causing the workqueue to requeue with an exponential backoff starting at
~5ms — effectively a tight retry loop that makes the rate-limit problem
worse.

The error comes back as *serviceerror.ResourceExhausted (not a standard
gRPC codes.ResourceExhausted status), so it must be detected with
errors.As rather than grpcstatus.FromError.

#### Changes

- internal/controller/worker_controller.go: detect
*serviceerror.ResourceExhausted from GetWorkerDeploymentState and return
RequeueAfter: 30s instead of an immediate error requeue. Sets
ConditionProgressing=False with ReasonTemporalStateFetchFailed and a
"Rate limited" message so the condition is visible to users.
- internal/tests/internal/rate_limit_integration_test.go: new
integration test that creates 10 TWDs against a 1 RPS limit, confirming
the error surfaces with the expected condition reason and message.
- go.mod / go.sum / go.work: bump Temporal server dependency to
v1.31.0-154.2, which is the first version that enforces
globalNamespaceWorkerDeploymentReadRPS.

## Checklist

1. Closes #278 

2. How was this tested:
- KUBEBUILDER_ASSETS=.../bin/k8s/1.27.1-darwin-arm64 go test -tags
test_dep ./internal/tests/internal -run "TestIntegration/rate-limit"
-timeout 120s -v passes
- Same test fails when the errors.As block is removed (reverts to
immediate retry with generic message)
- Full integration suite passes: go test -tags test_dep
./internal/tests/internal -run TestIntegration -timeout 600s

3. Any docs updates needed?
<!--- update README if applicable
      or point out where to update docs.temporal.io -->

Author: carlydf

internal/controller/worker_controller.go

@@ -14,6 +14,7 @@ import (
 	"github.com/temporalio/temporal-worker-controller/internal/controller/clientpool"
 	"github.com/temporalio/temporal-worker-controller/internal/k8s"
 	"github.com/temporalio/temporal-worker-controller/internal/temporal"
+	"go.temporal.io/api/serviceerror"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -242,6 +243,14 @@ func (r *TemporalWorkerDeploymentReconciler) Reconcile(ctx context.Context, req
 		getControllerIdentity(),
 	)
 	if err != nil {
+		var rateLimitErr *serviceerror.ResourceExhausted
+		if errors.As(err, &rateLimitErr) {
+			r.recordWarningAndSetBlocked(ctx, &workerDeploy,
+				temporaliov1alpha1.ReasonTemporalStateFetchFailed,
+				fmt.Sprintf("Rate limited fetching Temporal worker deployment state: %v", err),
+				fmt.Sprintf("Rate limited by Temporal server: %v", err))
+			return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
+		}
 		r.recordWarningAndSetBlocked(ctx, &workerDeploy,
 			temporaliov1alpha1.ReasonTemporalStateFetchFailed,
 			fmt.Sprintf("Unable to get Temporal worker deployment state: %v", err),

internal/tests/internal/integration_test.go

@@ -950,6 +950,17 @@ func TestIntegration(t *testing.T) {
 
 	// Conditions and events tests
 	runConditionsAndEventsTests(t, k8sClient, mgr, ts, testNamespace.Name)
+
+	// Rate limit test: uses a dedicated server to avoid interfering with the tests above.
+	// Ten TWDs in the same Temporal namespace produce 10 concurrent DescribeWorkerDeployment
+	// calls per second against a 1 RPS limit, reliably triggering ResourceExhausted errors.
+	dcRateLimit := dynamicconfig.NewMemoryClient()
+	dcRateLimit.OverrideValue(dynamicconfig.MakeKey("frontend.globalNamespaceWorkerDeploymentReadRPS"), 1)
+	tsRateLimit := temporaltest.NewServer(
+		temporaltest.WithT(t),
+		temporaltest.WithBaseServerOptions(temporal.WithDynamicConfigClient(dcRateLimit)),
+	)
+	runRateLimitTest(t, k8sClient, tsRateLimit, testNamespace.Name)
 }
 
 // testTemporalWorkerDeploymentCreation tests the creation of a TemporalWorkerDeployment and waits for the expected status

internal/tests/internal/rate_limit_integration_test.go

@@ -0,0 +1,95 @@
+package internal
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1"
+	"github.com/temporalio/temporal-worker-controller/internal/testhelpers"
+	"go.temporal.io/server/temporaltest"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// runRateLimitTest verifies that when multiple TWDs in the same Temporal namespace
+// burst-reconcile concurrently against a 1 RPS DescribeWorkerDeployment limit, the
+// controller surfaces the error as ConditionProgressing=False with
+// ReasonTemporalStateFetchFailed and a "Rate limited" message.
+//
+// The server passed in must have frontend.globalNamespaceWorkerDeploymentReadRPS=1.
+// With 10 TWDs each reconciling every 1s (RECONCILE_INTERVAL=1s set by setupTestEnvironment),
+// steady-state load is 10x the limit, ensuring the burst is reliably triggered.
+func runRateLimitTest(
+	t *testing.T,
+	k8sClient client.Client,
+	ts *temporaltest.TestServer,
+	testNamespace string,
+) {
+	t.Run("rate-limit-burst-surfaces-error", func(t *testing.T) {
+		ctx := context.Background()
+		const n = 10
+
+		twds := make([]*temporaliov1alpha1.TemporalWorkerDeployment, n)
+
+		for i := 0; i < n; i++ {
+			name := fmt.Sprintf("rate-limit-%d", i)
+
+			conn := &temporaliov1alpha1.TemporalConnection{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      name,
+					Namespace: testNamespace,
+				},
+				Spec: temporaliov1alpha1.TemporalConnectionSpec{
+					HostPort: ts.GetFrontendHostPort(),
+				},
+			}
+			if err := k8sClient.Create(ctx, conn); err != nil {
+				t.Fatalf("failed to create TemporalConnection %s: %v", name, err)
+			}
+
+			twd := testhelpers.NewTemporalWorkerDeploymentBuilder().
+				WithManualStrategy().
+				WithTargetTemplate("v1.0").
+				WithName(name).
+				WithNamespace(testNamespace).
+				WithTemporalConnection(name).
+				WithTemporalNamespace(ts.GetDefaultNamespace()).
+				Build()
+			if err := k8sClient.Create(ctx, twd); err != nil {
+				t.Fatalf("failed to create TWD %s: %v", name, err)
+			}
+			twds[i] = twd
+		}
+
+		// With 10 TWDs reconciling concurrently every 1s and a 1 RPS limit, most will be
+		// rate-limited almost immediately. Verify at least one surfaces the error with the
+		// expected condition reason and "Rate limited" message (set by our ResourceExhausted
+		// handler in worker_controller.go).
+		eventually(t, 30*time.Second, time.Second, func() error {
+			for _, twd := range twds {
+				var fresh temporaliov1alpha1.TemporalWorkerDeployment
+				if err := k8sClient.Get(ctx, types.NamespacedName{
+					Name:      twd.Name,
+					Namespace: twd.Namespace,
+				}, &fresh); err != nil {
+					return fmt.Errorf("get TWD %s: %w", twd.Name, err)
+				}
+				for _, c := range fresh.Status.Conditions {
+					if c.Type == temporaliov1alpha1.ConditionProgressing &&
+						c.Status == metav1.ConditionFalse &&
+						c.Reason == temporaliov1alpha1.ReasonTemporalStateFetchFailed &&
+						strings.Contains(c.Message, "Rate limited") {
+						t.Logf("TWD %s confirmed rate-limited: %s", twd.Name, c.Message)
+						return nil
+					}
+				}
+			}
+			return errors.New("no TWD has been rate-limited yet")
+		})
+	})
+}