omit DescribeVersion API calls for drained versions (#229)

## Summary
- Switch `GetWorkerDeploymentState` from the SDK's `Describe()` to the
gRPC `DescribeWorkerDeployment` API, which returns full `DrainageInfo`
(including `DrainedSince` timestamps) in the version summaries
- Eliminate per-version `DescribeVersion` calls for obtaining drainage
timestamps — read them directly from the gRPC summary
- Additionally, change/alter the constraint surrounding drained versions
that are considered eligible to be deleted by the server from the
worker-controller's pov.

## Test plan
- [ ] Verify existing unit tests pass (`go test
./internal/temporal/...`)
- [ ] Verify integration tests pass (`go test ./internal/tests/...`)
- [ ] Verify `DrainedSince` is correctly populated from the gRPC summary
- [ ] Verify `NoTaskQueuesHaveVersionedPoller` still works for drained
versions with k8s Deployments at 0 replicas
- [ ] Verify drained versions without k8s Deployments no longer trigger
`DescribeVersion` calls

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

---------

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

Author: Shivs11

api/v1alpha1/worker_types.go

@@ -308,9 +308,14 @@ type DeprecatedWorkerDeploymentVersion struct {
 	// +optional
 	DrainedSince *metav1.Time `json:"drainedSince,omitempty"`
 
-	// A Version is eligible for deletion if it is drained and has no pollers on any task queue.
-	// After pollers stop polling, the server will still consider them present until `matching.PollerHistoryTTL`
-	// has passed.
+	// A Version is considered eligible for deletion if it is drained and has no
+	// controller-managed workers with active replicas polling on its task queues.
+	// The server automatically deletes eligible versions when it needs to make room for new ones.
+	// If this version has pollers that are not managed by the controller, the server
+	// will not be able to delete the version. `EligibleForDeletion=true` does not consider
+	// that scenario, because it is rare and it is too expensive to cover (requires describing
+	// every drained version without controller-managed pollers and all task queues in
+	// that version).
 	// +optional
 	EligibleForDeletion bool `json:"eligibleForDeletion,omitempty"`
 }

internal/controller/reconciler_events_test.go

@@ -17,8 +17,11 @@ import (
 	temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1"
 	"github.com/temporalio/temporal-worker-controller/internal/controller/clientpool"
 	"github.com/temporalio/temporal-worker-controller/internal/planner"
+	deploymentpb "go.temporal.io/api/deployment/v1"
 	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/api/workflowservice/v1"
 	sdkclient "go.temporal.io/sdk/client"
+	"google.golang.org/grpc"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -215,24 +218,48 @@ type stubWDClient struct {
 
 func (s *stubWDClient) GetHandle(_ string) sdkclient.WorkerDeploymentHandle { return s.handle }
 
+// stubWorkflowServiceClient implements workflowservice.WorkflowServiceClient, returning
+// a valid empty response for DescribeWorkerDeployment (no versions, no routing config),
+// or a configurable error if describeDeploymentErr is set.
+type stubWorkflowServiceClient struct {
+	workflowservice.WorkflowServiceClient
+	describeDeploymentErr error
+}
+
+func (s *stubWorkflowServiceClient) DescribeWorkerDeployment(_ context.Context, _ *workflowservice.DescribeWorkerDeploymentRequest, _ ...grpc.CallOption) (*workflowservice.DescribeWorkerDeploymentResponse, error) {
+	if s.describeDeploymentErr != nil {
+		return nil, s.describeDeploymentErr
+	}
+	return &workflowservice.DescribeWorkerDeploymentResponse{
+		WorkerDeploymentInfo: &deploymentpb.WorkerDeploymentInfo{
+			RoutingConfig: &deploymentpb.RoutingConfig{},
+		},
+	}, nil
+}
+
 // stubTemporalClient implements sdkclient.Client, routing WorkerDeploymentClient and
 // ExecuteWorkflow to configurable stubs.
 type stubTemporalClient struct {
 	sdkclient.Client
-	wdClient sdkclient.WorkerDeploymentClient
-	execErr  error
+	wdClient              sdkclient.WorkerDeploymentClient
+	execErr               error
+	describeDeploymentErr error
 }
 
 func (s *stubTemporalClient) WorkerDeploymentClient() sdkclient.WorkerDeploymentClient {
 	return s.wdClient
 }
 
+func (s *stubTemporalClient) WorkflowService() workflowservice.WorkflowServiceClient {
+	return &stubWorkflowServiceClient{describeDeploymentErr: s.describeDeploymentErr}
+}
+
 func (s *stubTemporalClient) ExecuteWorkflow(_ context.Context, _ sdkclient.StartWorkflowOptions, _ interface{}, _ ...interface{}) (sdkclient.WorkflowRun, error) {
 	return nil, s.execErr
 }
 
-// newStubTemporalClient returns a stub client whose Describe returns NotFound (no existing
-// Worker Deployment), and whose ExecuteWorkflow returns execErr.
+// newStubTemporalClient returns a stub client whose WorkflowService().DescribeWorkerDeployment
+// returns a valid empty response, and whose ExecuteWorkflow returns execErr.
 func newStubTemporalClient(execErr error) *stubTemporalClient {
 	handle := &stubWDHandle{describeErr: &serviceerror.NotFound{}}
 	return &stubTemporalClient{
@@ -511,6 +538,36 @@ func TestReconcile_PlanExecutionFailed_EmitsEvent(t *testing.T) {
 	assertEventEmitted(t, drainEvents(recorder), ReasonPlanExecutionFailed)
 }
 
+// TestReconcile_DescribeWorkerDeploymentNotFound verifies that when the gRPC
+// DescribeWorkerDeployment call returns NotFound (no deployment exists in Temporal yet),
+// reconciliation succeeds and proceeds to plan generation (creating a new k8s Deployment).
+func TestReconcile_DescribeWorkerDeploymentNotFound(t *testing.T) {
+	k8sNamespace := "default"
+	hostPort := "localhost:7233"
+
+	tc := makeNoCredsTemporalConnection("my-conn", k8sNamespace, hostPort)
+	twd := makeTWD("test-worker", k8sNamespace, tc.Name)
+
+	r, recorder := newTestReconcilerWithInterceptors([]client.Object{twd, tc}, interceptor.Funcs{})
+
+	stub := newStubTemporalClient(nil)
+	stub.describeDeploymentErr = &serviceerror.NotFound{}
+	r.TemporalClientPool.SetClientForTesting(
+		noCredsPoolKey(tc.Spec.HostPort, twd.Spec.WorkerOptions.TemporalNamespace),
+		stub,
+	)
+
+	_, err := r.Reconcile(context.Background(), ctrl.Request{
+		NamespacedName: types.NamespacedName{Name: twd.Name, Namespace: twd.Namespace},
+	})
+
+	// NotFound on the first reconcile means the Worker Deployment has not come up on the
+	// server side yet; however, the k8s Deployment would be created by the controller
+	// with no reconciliation errors.
+	require.NoError(t, err)
+	assertNoEventEmitted(t, drainEvents(recorder), ReasonPlanGenerationFailed)
+}
+
 // ─── executeK8sOperations tests ──────────────────────────────────────────────
 
 func TestExecuteK8sOperations_EmitsEventOnFailure(t *testing.T) {

internal/controller/state_mapper.go

@@ -157,9 +157,19 @@ func (m *stateMapper) mapDeprecatedWorkerDeploymentVersionByBuildID(buildID stri
 		return nil
 	}
 
+	// A drained version is eligible for server-side auto-deletion only if its pollers
+	// have stopped. We approximate this by checking whether a controller-managed k8s
+	// Deployment with active replicas exists; if so, pollers are likely still running
+	// and the server cannot delete the version yet, making it ineligible for deletion.
+	// Note: This only considers pollers managed by the worker-controller. External
+	// pollers (e.g., manually deployed workers) are not accounted for.
 	eligibleForDeletion := false
 	if vInfo, exists := m.temporalState.Versions[buildID]; exists {
-		eligibleForDeletion = vInfo.Status == v1alpha1.VersionStatusDrained && vInfo.NoTaskQueuesHaveVersionedPoller
+		hasActiveDeployment := false
+		if d, ok := m.k8sState.Deployments[buildID]; ok {
+			hasActiveDeployment = d.Status.Replicas > 0
+		}
+		eligibleForDeletion = vInfo.Status == v1alpha1.VersionStatusDrained && !hasActiveDeployment
 	}
 
 	version := &v1alpha1.DeprecatedWorkerDeploymentVersion{