New matching RPC: Check if version exists in TaskQueue

common/worker_versioning/worker_versioning.go

@@ -265,21 +265,19 @@ func GetIsWFTaskQueueInVersionDetector(matchingClient resource.MatchingClient) I
 	return func(ctx context.Context,
 		namespaceID, tq string,
 		version *deploymentpb.WorkerDeploymentVersion) (bool, error) {
-		resp, err := matchingClient.GetTaskQueueUserData(ctx,
-			&matchingservice.GetTaskQueueUserDataRequest{
-				NamespaceId:   namespaceID,
-				TaskQueue:     tq,
-				TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW,
-			})
+		resp, err := matchingClient.CheckTaskQueueVersionMembership(ctx, &matchingservice.CheckTaskQueueVersionMembershipRequest{
+			NamespaceId:   namespaceID,
+			TaskQueue:     tq,
+			TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW,
+			Version: &deploymentspb.WorkerDeploymentVersion{
+				DeploymentName: version.GetDeploymentName(),
+				BuildId:        version.GetBuildId(),
+			},
+		})
 		if err != nil {
 			return false, err
 		}
-		tqData, ok := resp.GetUserData().GetData().GetPerType()[int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW)]
-		if !ok {
-			// The TQ is unversioned
-			return false, nil
-		}
-		return HasDeploymentVersion(tqData.GetDeploymentData(), DeploymentVersionFromDeployment(DeploymentFromExternalDeploymentVersion(version))), nil
+		return resp.GetIsMember(), nil
 	}
 }
 

proto/internal/temporal/server/api/matchingservice/v1/request_response.proto

@@ -641,3 +641,14 @@ message DescribeWorkerRequest {
 message DescribeWorkerResponse {
     temporal.api.worker.v1.WorkerInfo worker_info = 1;
 }
+
+message CheckTaskQueueVersionMembershipRequest {
+    string namespace_id = 1;
+    string task_queue = 2;
+    temporal.api.enums.v1.TaskQueueType task_queue_type = 3;
+    temporal.server.api.deployment.v1.WorkerDeploymentVersion version = 4;
+}
+
+message CheckTaskQueueVersionMembershipResponse {
+    bool is_member = 1;
+}

proto/internal/temporal/server/api/matchingservice/v1/service.proto

@@ -190,5 +190,8 @@ service MatchingService {
     // Returns an error if the namespace or worker doesn't exist.
     rpc DescribeWorker (DescribeWorkerRequest) returns (DescribeWorkerResponse) {}
 
+    // CheckTaskQueueVersionMembership checks if a task queue is part of a specific deployment version.
+    rpc CheckTaskQueueVersionMembership(CheckTaskQueueVersionMembershipRequest) returns (CheckTaskQueueVersionMembershipResponse) {}
+
 }
 

service/matching/handler.go

@@ -475,6 +475,14 @@ func (h *Handler) CheckTaskQueueUserDataPropagation(
 	return h.engine.CheckTaskQueueUserDataPropagation(ctx, request)
 }
 
+func (h *Handler) CheckTaskQueueVersionMembership(
+	ctx context.Context,
+	request *matchingservice.CheckTaskQueueVersionMembershipRequest,
+) (_ *matchingservice.CheckTaskQueueVersionMembershipResponse, retError error) {
+	defer log.CapturePanic(h.logger, &retError)
+	return h.engine.CheckTaskQueueVersionMembership(ctx, request)
+}
+
 func (h *Handler) DispatchNexusTask(ctx context.Context, request *matchingservice.DispatchNexusTaskRequest) (_ *matchingservice.DispatchNexusTaskResponse, retError error) {
 	defer log.CapturePanic(h.logger, &retError)
 	return h.engine.DispatchNexusTask(ctx, request)

service/matching/matching_engine.go

@@ -3139,6 +3139,32 @@ func prepareTaskQueueUserData(
 	return data
 }
 
+func (e *matchingEngineImpl) CheckTaskQueueVersionMembership(
+	ctx context.Context,
+	request *matchingservice.CheckTaskQueueVersionMembershipRequest,
+) (*matchingservice.CheckTaskQueueVersionMembershipResponse, error) {
+	partition, err := tqid.PartitionFromProto(&taskqueuepb.TaskQueue{
+		Name: request.GetTaskQueue(),
+		Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
+	}, request.GetNamespaceId(), request.GetTaskQueueType())
+	if err != nil {
+		return nil, err
+	}
+	pm, _, err := e.getTaskQueuePartitionManager(ctx, partition, true, loadCauseOtherRead)
+	if err != nil {
+		return nil, err
+	}
+
+	userData, _, err := pm.GetUserDataManager().GetUserData()
+	if err != nil {
+		return nil, err
+	}
+
+	typedUserData := userData.GetData().GetPerType()[int32(request.GetTaskQueueType())]
+	present := worker_versioning.HasDeploymentVersion(typedUserData.GetDeploymentData(), request.GetVersion())
+	return &matchingservice.CheckTaskQueueVersionMembershipResponse{IsMember: present}, nil
+}
+
 func (e *matchingEngineImpl) UpdateTaskQueueConfig(
 	ctx context.Context,
 	request *matchingservice.UpdateTaskQueueConfigRequest,