Add WorkerCommandsPartition type and partition metrics tag

When the SDK sets Kind=TASK_QUEUE_KIND_WORKER_COMMANDS, the server now
creates a WorkerCommandsPartition (similar to StickyPartition). Metrics
for these partitions get partition=__worker_commands__ and the taskqueue
tag is suppressed to __omitted__ to prevent cardinality explosion from
per-worker queue names.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: rkannan82

common/metrics/task_queues.go

@@ -7,9 +7,10 @@ import (
 )
 
 const (
-	omitted = "__omitted__"
-	normal  = "__normal__"
-	sticky  = "__sticky__"
+	omitted        = "__omitted__"
+	normal         = "__normal__"
+	sticky         = "__sticky__"
+	workerCommands = "__worker_commands__"
 )
 
 // GetPerTaskQueueFamilyScope returns "namespace" and "taskqueue" tags. "taskqueue" will be "__omitted__" if
@@ -53,15 +54,19 @@ func GetPerTaskQueuePartitionIDScope(
 	tags ...Tag,
 ) Handler {
 	var value string
-	if partition == nil {
+	switch p := partition.(type) {
+	case nil:
 		value = unknownValue
-	} else if normalPartition, ok := partition.(*tqid.NormalPartition); ok {
+	case *tqid.NormalPartition:
 		if partitionIDBreakdown {
-			value = strconv.Itoa(normalPartition.PartitionId())
+			value = strconv.Itoa(p.PartitionId())
 		} else {
 			value = normal
 		}
-	} else {
+	case *tqid.WorkerCommandsPartition:
+		value = workerCommands
+		taskQueueBreakdown = false
+	default:
 		value = sticky
 	}
 
@@ -70,7 +75,7 @@ func GetPerTaskQueuePartitionIDScope(
 }
 
 // GetPerTaskQueuePartitionTypeScope returns GetPerTaskQueueScope scope plus a "partition" tag which
-// can be "__normal__", "__sticky__", or "_unknown_".
+// can be "__normal__", "__sticky__", "__worker_commands__", or "_unknown_".
 func GetPerTaskQueuePartitionTypeScope(
 	handler Handler,
 	namespaceName string,
@@ -79,11 +84,15 @@ func GetPerTaskQueuePartitionTypeScope(
 	tags ...Tag,
 ) Handler {
 	var value string
-	if partition == nil {
+	switch partition.(type) {
+	case nil:
 		value = unknownValue
-	} else if _, ok := partition.(*tqid.NormalPartition); ok {
+	case *tqid.NormalPartition:
 		value = normal
-	} else {
+	case *tqid.WorkerCommandsPartition:
+		value = workerCommands
+		taskQueueBreakdown = false
+	default:
 		value = sticky
 	}
 

common/tqid/task_queue_id.go

@@ -88,6 +88,14 @@ type (
 		taskQueue  *TaskQueue
 	}
 
+	// WorkerCommandsPartition is used for server-to-worker communication (e.g. activity cancellations).
+	// These queues are per-worker and only exist for the lifetime of the worker process. The SDK sets
+	// Kind=TASK_QUEUE_KIND_WORKER_COMMANDS when polling on these queues.
+	WorkerCommandsPartition struct {
+		name      string
+		taskQueue *TaskQueue
+	}
+
 	// PartitionKey uniquely identifies a task queue partition, to be used in maps.
 	// Note that task queue kind (sticky vs normal) and normal name for sticky task queues are not
 	// part of the task queue partition identity.
@@ -101,6 +109,7 @@ type (
 
 var _ Partition = (*NormalPartition)(nil)
 var _ Partition = (*StickyPartition)(nil)
+var _ Partition = (*WorkerCommandsPartition)(nil)
 
 var (
 	ErrNoParent      = errors.New("root task queue partition has no parent")
@@ -139,6 +148,9 @@ func UnsafePartitionFromProto(proto *taskqueuepb.TaskQueue, namespaceId string,
 	case enumspb.TASK_QUEUE_KIND_STICKY:
 		tq := &TaskQueue{TaskQueueFamily{namespaceId, proto.GetNormalName()}, taskType}
 		return tq.StickyPartition(proto.GetName())
+	case enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS:
+		tq := &TaskQueue{TaskQueueFamily{namespaceId, proto.GetName()}, taskType}
+		return tq.WorkerCommandsPartition(proto.GetName())
 	default:
 		tq := &TaskQueue{TaskQueueFamily{namespaceId, proto.GetName()}, taskType}
 		return tq.RootPartition()
@@ -164,6 +176,12 @@ func PartitionFromProto(proto *taskqueuepb.TaskQueue, namespaceId string, taskTy
 		}
 		tq := &TaskQueue{TaskQueueFamily{namespaceId, normalName}, taskType}
 		return tq.StickyPartition(baseName), nil
+	case enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS:
+		if partition != 0 {
+			return nil, serviceerror.NewInvalidArgumentf("worker-commands partitions cannot have non-zero partition ID. base name: %s", baseName)
+		}
+		tq := &TaskQueue{TaskQueueFamily{namespaceId, baseName}, taskType}
+		return tq.WorkerCommandsPartition(baseName), nil
 	default:
 		tq := &TaskQueue{TaskQueueFamily{namespaceId, baseName}, taskType}
 		return tq.NormalPartition(partition), nil
@@ -239,6 +257,10 @@ func (n *TaskQueue) StickyPartition(stickyName string) *StickyPartition {
 	return &StickyPartition{stickyName, n}
 }
 
+func (n *TaskQueue) WorkerCommandsPartition(name string) *WorkerCommandsPartition {
+	return &WorkerCommandsPartition{name, n}
+}
+
 func (n *TaskQueue) RootPartition() *NormalPartition {
 	return n.NormalPartition(0)
 }
@@ -300,6 +322,51 @@ func (s *StickyPartition) GradualChangeKey() []byte {
 	return []byte(key)
 }
 
+func (w *WorkerCommandsPartition) TaskType() enumspb.TaskQueueType {
+	return w.taskQueue.TaskType()
+}
+
+func (w *WorkerCommandsPartition) Kind() enumspb.TaskQueueKind {
+	return enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS
+}
+
+func (w *WorkerCommandsPartition) NamespaceId() string {
+	return w.taskQueue.family.NamespaceId()
+}
+
+func (w *WorkerCommandsPartition) TaskQueue() *TaskQueue {
+	return w.taskQueue
+}
+
+func (w *WorkerCommandsPartition) IsRoot() bool {
+	return false
+}
+
+func (w *WorkerCommandsPartition) IsChild() bool {
+	return false
+}
+
+func (w *WorkerCommandsPartition) RpcName() string {
+	return w.name
+}
+
+func (w *WorkerCommandsPartition) Key() PartitionKey {
+	return PartitionKey{
+		namespaceId: w.NamespaceId(),
+		name:        w.name,
+		taskType:    w.TaskType(),
+	}
+}
+
+func (w *WorkerCommandsPartition) RoutingKey(int) (string, int) {
+	return fmt.Sprintf("%s:%s:%d", w.NamespaceId(), w.RpcName(), w.TaskType()), 0
+}
+
+func (w *WorkerCommandsPartition) GradualChangeKey() []byte {
+	key := fmt.Sprintf("%s:%s:%d", w.NamespaceId(), w.RpcName(), w.TaskType())
+	return []byte(key)
+}
+
 func (p *NormalPartition) TaskQueue() *TaskQueue {
 	return p.taskQueue
 }

service/matching/matching_engine_test.go

@@ -5232,6 +5232,131 @@ func (*testTaskManager) CountTaskQueuesByBuildId(context.Context, *persistence.C
 	return 0, nil
 }
 
+// TestLoggerAndMetricsForPartition_BreakdownEnabled verifies partition and taskqueue tag values
+// with the default BreakdownMetricsByTaskQueue=true: normal queues show their name, sticky queues
+// show the base name, and worker-commands queues get partition=__worker_commands__ with
+// taskqueue=__omitted__.
+func TestLoggerAndMetricsForPartition_BreakdownEnabled(t *testing.T) {
+	t.Parallel()
+
+	controller := gomock.NewController(t)
+	ns, mockNamespaceCache := createMockNamespaceCache(controller, matchingTestNamespace)
+	config := defaultTestConfig()
+	e := createTestMatchingEngine(log.NewTestLogger(), controller, config, nil, mockNamespaceCache)
+	captureHandler := metricstest.NewCaptureHandler()
+	e.metricsHandler = captureHandler
+
+	tests := []struct {
+		name               string
+		partition          tqid.Partition
+		expectTQValue      string
+		expectPartitionTag string
+	}{
+		{
+			name:               "normal task queue uses actual queue name",
+			partition:          newRootPartition(ns.ID().String(), "my-task-queue", enumspb.TASK_QUEUE_TYPE_NEXUS),
+			expectTQValue:      "my-task-queue",
+			expectPartitionTag: "0",
+		},
+		{
+			name:               "worker-commands queue gets __worker_commands__ partition and __omitted__ taskqueue",
+			partition:          newTestTaskQueue(ns.ID().String(), "my-task-queue", enumspb.TASK_QUEUE_TYPE_NEXUS).WorkerCommandsPartition("/temporal-sys/worker-commands/ns/key"),
+			expectTQValue:      "__omitted__",
+			expectPartitionTag: "__worker_commands__",
+		},
+		{
+			name:               "sticky task queue uses base queue name",
+			partition:          newTestTaskQueue(ns.ID().String(), "my-task-queue", enumspb.TASK_QUEUE_TYPE_WORKFLOW).StickyPartition(uuid.NewString()),
+			expectTQValue:      "my-task-queue",
+			expectPartitionTag: "__sticky__",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			capture := captureHandler.StartCapture()
+			tqConfig := newTaskQueueConfig(tc.partition.TaskQueue(), config, matchingTestNamespace)
+			_, _, handler := e.loggerAndMetricsForPartition(ns, tc.partition, tqConfig)
+			metrics.PollSuccessPerTaskQueueCounter.With(handler).Record(1)
+			snap := capture.Snapshot()
+			captureHandler.StopCapture(capture)
+			recordings := snap["poll_success"]
+			require.NotEmpty(t, recordings, "expected poll_success metric to be recorded")
+			found := false
+			for _, rec := range recordings {
+				if rec.Tags["taskqueue"] == tc.expectTQValue && rec.Tags["partition"] == tc.expectPartitionTag {
+					found = true
+				}
+			}
+			assert.True(t, found, "expected taskqueue=%q partition=%q, got: %v", tc.expectTQValue, tc.expectPartitionTag, recordings)
+		})
+	}
+}
+
+// TestLoggerAndMetricsForPartition_BreakdownDisabled verifies behavior with BreakdownMetricsByTaskQueue=false:
+// normal and sticky queues get taskqueue=__omitted__, and worker-commands queues also get __omitted__
+// with partition=__worker_commands__.
+func TestLoggerAndMetricsForPartition_BreakdownDisabled(t *testing.T) {
+	t.Parallel()
+
+	controller := gomock.NewController(t)
+	ns, mockNamespaceCache := createMockNamespaceCache(controller, matchingTestNamespace)
+	dc := dynamicconfig.StaticClient{
+		dynamicconfig.MetricsBreakdownByTaskQueue.Key(): false,
+	}
+	config := NewConfig(dynamicconfig.NewCollection(dc, log.NewNoopLogger()))
+	config.LongPollExpirationInterval = dynamicconfig.GetDurationPropertyFnFilteredByTaskQueue(100 * time.Millisecond)
+	e := createTestMatchingEngine(log.NewTestLogger(), controller, config, nil, mockNamespaceCache)
+	captureHandler := metricstest.NewCaptureHandler()
+	e.metricsHandler = captureHandler
+
+	tests := []struct {
+		name               string
+		partition          tqid.Partition
+		expectTQValue      string
+		expectPartitionTag string
+	}{
+		{
+			name:               "normal task queue is omitted when breakdown disabled",
+			partition:          newRootPartition(ns.ID().String(), "my-task-queue", enumspb.TASK_QUEUE_TYPE_NEXUS),
+			expectTQValue:      "__omitted__",
+			expectPartitionTag: "0",
+		},
+		{
+			name:               "worker-commands queue gets __worker_commands__ partition when breakdown disabled",
+			partition:          newTestTaskQueue(ns.ID().String(), "my-task-queue", enumspb.TASK_QUEUE_TYPE_NEXUS).WorkerCommandsPartition("/temporal-sys/worker-commands/ns/key"),
+			expectTQValue:      "__omitted__",
+			expectPartitionTag: "__worker_commands__",
+		},
+		{
+			name:               "sticky task queue is omitted when breakdown disabled",
+			partition:          newTestTaskQueue(ns.ID().String(), "my-task-queue", enumspb.TASK_QUEUE_TYPE_WORKFLOW).StickyPartition(uuid.NewString()),
+			expectTQValue:      "__omitted__",
+			expectPartitionTag: "__sticky__",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			capture := captureHandler.StartCapture()
+			tqConfig := newTaskQueueConfig(tc.partition.TaskQueue(), config, matchingTestNamespace)
+			_, _, handler := e.loggerAndMetricsForPartition(ns, tc.partition, tqConfig)
+			metrics.PollSuccessPerTaskQueueCounter.With(handler).Record(1)
+			snap := capture.Snapshot()
+			captureHandler.StopCapture(capture)
+			recordings := snap["poll_success"]
+			require.NotEmpty(t, recordings, "expected poll_success metric to be recorded")
+			found := false
+			for _, rec := range recordings {
+				if rec.Tags["taskqueue"] == tc.expectTQValue && rec.Tags["partition"] == tc.expectPartitionTag {
+					found = true
+				}
+			}
+			assert.True(t, found, "expected taskqueue=%q partition=%q, got: %v", tc.expectTQValue, tc.expectPartitionTag, recordings)
+		})
+	}
+}
+
 func TestConvertPollWorkflowTaskQueueResponse(t *testing.T) {
 	t.Parallel()