Matching logic for pinned workflows

common/worker_versioning/worker_versioning.go

@@ -33,6 +33,7 @@ import (
 	commonpb "go.temporal.io/api/common/v1"
 	deploymentpb "go.temporal.io/api/deployment/v1"
 	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/api/serviceerror"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
 	"go.temporal.io/server/common/namespace"
@@ -222,3 +223,18 @@ func StampFromCapabilities(cap *commonpb.WorkerVersionCapabilities) *commonpb.Wo
 func StampFromBuildId(buildId string) *commonpb.WorkerVersionStamp {
 	return &commonpb.WorkerVersionStamp{UseVersioning: true, BuildId: buildId}
 }
+
+// ValidateDeployment returns error if the deployment is nil or it has empty build ID or deployment
+// name.
+func ValidateDeployment(deployment *deploymentpb.Deployment) error {
+	if deployment == nil {
+		return serviceerror.NewInvalidArgument("deployment cannot be nil")
+	}
+	if deployment.GetSeriesName() == "" {
+		return serviceerror.NewInvalidArgument("deployment series name cannot be empty")
+	}
+	if deployment.GetBuildId() == "" {
+		return serviceerror.NewInvalidArgument("deployment build ID cannot be empty")
+	}
+	return nil
+}

service/matching/db.go

@@ -222,7 +222,7 @@ func (db *taskQueueDB) updateApproximateBacklogCount(
 	// Prevent under-counting
 	if db.approximateBacklogCount.Load()+delta < 0 {
 		db.logger.Info("ApproximateBacklogCounter could have under-counted.",
-			tag.WorkerBuildId(db.queue.BuildId()), tag.WorkflowNamespace(db.queue.Partition().NamespaceId()))
+			tag.WorkerBuildId(db.queue.Version().MetricsTagValue()), tag.WorkflowNamespace(db.queue.Partition().NamespaceId()))
 		db.approximateBacklogCount.Store(0)
 	} else {
 		db.approximateBacklogCount.Add(delta)

service/matching/forwarder.go

@@ -190,8 +190,8 @@ func (fwdr *Forwarder) getForwardInfo(task *internalTask) *taskqueuespb.TaskForw
 	forwardInfo := &taskqueuespb.TaskForwardInfo{
 		TaskSource:         task.source,
 		SourcePartition:    fwdr.partition.RpcName(),
-		DispatchBuildId:    fwdr.queue.BuildId(),
-		DispatchVersionSet: fwdr.queue.VersionSet(),
+		DispatchBuildId:    fwdr.queue.Version().BuildId(),
+		DispatchVersionSet: fwdr.queue.Version().VersionSet(),
 		RedirectInfo:       task.redirectInfo,
 	}
 	return forwardInfo

service/matching/matching_engine.go

@@ -504,7 +504,6 @@ func (e *matchingEngineImpl) AddWorkflowTask(
 
 	return pm.AddTask(ctx, addTaskParams{
 		taskInfo:    taskInfo,
-		directive:   addRequest.VersionDirective,
 		forwardInfo: addRequest.ForwardInfo,
 	})
 }
@@ -542,7 +541,6 @@ func (e *matchingEngineImpl) AddActivityTask(
 
 	return pm.AddTask(ctx, addTaskParams{
 		taskInfo:    taskInfo,
-		directive:   addRequest.VersionDirective,
 		forwardInfo: addRequest.ForwardInfo,
 	})
 }
@@ -2128,9 +2126,11 @@ func (e *matchingEngineImpl) unloadTaskQueuePartitionByKey(
 func (e *matchingEngineImpl) updatePhysicalTaskQueueGauge(pqm *physicalTaskQueueManagerImpl, delta int) {
 	// calculating versioned to be one of: “unversioned” or "buildId” or “versionSet”
 	versioned := "unversioned"
-	if buildID := pqm.queue.BuildId(); buildID != "" {
+	if dep := pqm.queue.Version().Deployment(); dep != nil {
+		versioned = "deployment"
+	} else if buildID := pqm.queue.Version().BuildId(); buildID != "" {
 		versioned = "buildId"
-	} else if versionSet := pqm.queue.VersionSet(); versionSet != "" {
+	} else if versionSet := pqm.queue.Version().VersionSet(); versionSet != "" {
 		versioned = "versionSet"
 	}
 

service/matching/matching_engine_test.go

@@ -2970,13 +2970,18 @@ func (s *matchingEngineSuite) TestUpdatePhysicalTaskQueueGauge_VersionSet() {
 	// the size of the map to 1 and it's counter to 1.
 	s.PhysicalQueueMetricValidator(capture, 1, 1)
 
-	Vqtpm, err := tqm.(*taskQueuePartitionManagerImpl).getVersionedQueueNoWait(versionSet, "", true)
+	vqtpm, err := tqm.(*taskQueuePartitionManagerImpl).getVersionedQueueNoWait(
+		versionSet,
+		"",
+		nil,
+		true,
+	)
 	s.Require().NoError(err)
 
 	// Creating a VersionedQueue results in increasing the size of the map to 2, due to 2 entries now,
 	// with it's counter to 1.
 	s.PhysicalQueueMetricValidator(capture, 2, 1)
-	s.matchingEngine.updatePhysicalTaskQueueGauge(Vqtpm.(*physicalTaskQueueManagerImpl), 1)
+	s.matchingEngine.updatePhysicalTaskQueueGauge(vqtpm.(*physicalTaskQueueManagerImpl), 1)
 	s.PhysicalQueueMetricValidator(capture, 3, 2)
 
 	// Validating if versioned has been set right for the specific parameters
@@ -3007,7 +3012,12 @@ func (s *matchingEngineSuite) TestUpdatePhysicalTaskQueueGauge_BuildID() {
 	// the size of the map to 1 and it's counter to 1.
 	s.PhysicalQueueMetricValidator(capture, 1, 1)
 
-	Vqtpm, err := tqm.(*taskQueuePartitionManagerImpl).getVersionedQueueNoWait("", buildID, true)
+	Vqtpm, err := tqm.(*taskQueuePartitionManagerImpl).getVersionedQueueNoWait(
+		"",
+		buildID,
+		nil,
+		true,
+	)
 	s.Require().NoError(err)
 
 	// Creating a VersionedQueue results in increasing the size of the map to 2, due to 2 entries now,
@@ -3473,12 +3483,11 @@ type testTaskManager struct {
 
 type dbTaskQueueKey struct {
 	partitionKey tqid.PartitionKey
-	versionSet   string
-	buildId      string
+	version      PhysicalTaskQueueVersion
 }
 
 func getKey(dbq *PhysicalTaskQueueKey) dbTaskQueueKey {
-	return dbTaskQueueKey{dbq.partition.Key(), dbq.versionSet, dbq.buildId}
+	return dbTaskQueueKey{dbq.partition.Key(), dbq.Version()}
 }
 
 func newTestTaskManager(logger log.Logger) *testTaskManager {

service/matching/physical_task_queue_key.go

@@ -31,16 +31,18 @@ import (
 	"strconv"
 	"strings"
 
+	"go.temporal.io/api/deployment/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/server/common/tqid"
 )
 
 const (
 	// nonRootPartitionPrefix is the prefix for all mangled task queue names.
-	nonRootPartitionPrefix = "/_sys/"
-	partitionDelimiter     = "/"
-	versionSetDelimiter    = ":"
-	buildIdDelimiter       = "#"
+	nonRootPartitionPrefix  = "/_sys/"
+	partitionDelimiter      = "/"
+	versionSetDelimiter     = ":"
+	buildIdDelimiter        = "#"
+	deploymentNameDelimiter = "|"
 )
 
 type (
@@ -51,10 +53,16 @@ type (
 	// Physical task queues with a version set or build ID are called "versioned". The ones without a version set
 	// or build ID are called "unversioned". A physical queue cannot have both version set and build ID.
 	PhysicalTaskQueueKey struct {
-		partition  tqid.Partition
+		partition tqid.Partition
+		version   PhysicalTaskQueueVersion
+	}
+
+	PhysicalTaskQueueVersion struct {
 		versionSet string // version set id
-		// BuildId and VersionSet are mutually exclusive
+		// buildId and versionSet are mutually exclusive. buildId must be set if deploymentSeriesName is.
 		buildId string
+		// When present, it means this is a V3 pinned queue.
+		deploymentSeriesName string
 	}
 )
 
@@ -78,14 +86,6 @@ func (q *PhysicalTaskQueueKey) Partition() tqid.Partition {
 	return q.partition
 }
 
-func (q *PhysicalTaskQueueKey) BuildId() string {
-	return q.buildId
-}
-
-func (q *PhysicalTaskQueueKey) VersionSet() string {
-	return q.versionSet
-}
-
 // UnversionedQueueKey returns the unversioned PhysicalTaskQueueKey of a task queue partition
 func UnversionedQueueKey(p tqid.Partition) *PhysicalTaskQueueKey {
 	return &PhysicalTaskQueueKey{
@@ -96,16 +96,31 @@ func UnversionedQueueKey(p tqid.Partition) *PhysicalTaskQueueKey {
 // VersionSetQueueKey returns a PhysicalTaskQueueKey of a task queue partition with the given version set id.
 func VersionSetQueueKey(p tqid.Partition, versionSet string) *PhysicalTaskQueueKey {
 	return &PhysicalTaskQueueKey{
-		partition:  p,
-		versionSet: versionSet,
+		partition: p,
+		version: PhysicalTaskQueueVersion{
+			versionSet: versionSet,
+		},
 	}
 }
 
 // BuildIdQueueKey returns a PhysicalTaskQueueKey of a task queue partition with the given build ID.
 func BuildIdQueueKey(p tqid.Partition, buildId string) *PhysicalTaskQueueKey {
 	return &PhysicalTaskQueueKey{
 		partition: p,
-		buildId:   buildId,
+		version: PhysicalTaskQueueVersion{
+			buildId: buildId,
+		},
+	}
+}
+
+// DeploymentQueueKey returns a PhysicalTaskQueueKey of a task queue partition for a deployment.
+func DeploymentQueueKey(p tqid.Partition, deployment *deployment.Deployment) *PhysicalTaskQueueKey {
+	return &PhysicalTaskQueueKey{
+		partition: p,
+		version: PhysicalTaskQueueVersion{
+			buildId:              deployment.GetBuildId(),
+			deploymentSeriesName: deployment.GetSeriesName(),
+		},
 	}
 }
 
@@ -119,6 +134,7 @@ func BuildIdQueueKey(p tqid.Partition, buildId string) *PhysicalTaskQueueKey {
 //
 // All versioned DB queues use mangled names, using the following format:
 //
+//	with build ID: 		/_sys/<base name>/<deployment name base64 URL encoded>|<build ID base64 URL encoded>#<partition id>
 //	with build ID: 		/_sys/<base name>/<build ID base64 URL encoded>#<partition id>
 //	with version set: 		/_sys/<base name>/<version set id>:<partition id>
 func (q *PhysicalTaskQueueKey) PersistenceName() string {
@@ -128,12 +144,16 @@ func (q *PhysicalTaskQueueKey) PersistenceName() string {
 	case *tqid.NormalPartition:
 		baseName := q.TaskQueueFamily().Name()
 
-		if len(q.versionSet) > 0 {
-			return nonRootPartitionPrefix + baseName + partitionDelimiter + q.versionSet + versionSetDelimiter + strconv.Itoa(p.PartitionId())
+		if len(q.version.versionSet) > 0 {
+			return nonRootPartitionPrefix + baseName + partitionDelimiter + q.version.versionSet + versionSetDelimiter + strconv.Itoa(p.PartitionId())
 		}
 
-		if len(q.buildId) > 0 {
-			encodedBuildId := base64.URLEncoding.EncodeToString([]byte(q.buildId))
+		if len(q.version.deploymentSeriesName) > 0 {
+			encodedBuildId := base64.RawURLEncoding.EncodeToString([]byte(q.version.buildId))
+			encodedDeploymentName := base64.RawURLEncoding.EncodeToString([]byte(q.version.deploymentSeriesName))
+			return nonRootPartitionPrefix + baseName + partitionDelimiter + encodedDeploymentName + deploymentNameDelimiter + encodedBuildId + buildIdDelimiter + strconv.Itoa(p.PartitionId())
+		} else if len(q.version.buildId) > 0 {
+			encodedBuildId := base64.URLEncoding.EncodeToString([]byte(q.version.buildId))
 			return nonRootPartitionPrefix + baseName + partitionDelimiter + encodedBuildId + buildIdDelimiter + strconv.Itoa(p.PartitionId())
 		}
 
@@ -154,6 +174,7 @@ func ParsePhysicalTaskQueueKey(persistenceName string, namespaceId string, taskT
 	partitionId := 0
 	versionSet := ""
 	buildId := ""
+	deploymentName := ""
 
 	if strings.HasPrefix(persistenceName, nonRootPartitionPrefix) {
 		suffixOff := strings.LastIndex(persistenceName, partitionDelimiter)
@@ -163,7 +184,7 @@ func ParsePhysicalTaskQueueKey(persistenceName string, namespaceId string, taskT
 		baseName = persistenceName[len(nonRootPartitionPrefix):suffixOff]
 		suffix := persistenceName[suffixOff+1:]
 		var err error
-		partitionId, versionSet, buildId, err = parseSuffix(persistenceName, suffix)
+		partitionId, versionSet, buildId, deploymentName, err = parseSuffix(persistenceName, suffix)
 		if err != nil {
 			return nil, err
 		}
@@ -174,36 +195,102 @@ func ParsePhysicalTaskQueueKey(persistenceName string, namespaceId string, taskT
 		return nil, err
 	}
 	return &PhysicalTaskQueueKey{
-		partition:  f.TaskQueue(taskType).NormalPartition(partitionId),
-		versionSet: versionSet,
-		buildId:    buildId,
+		partition: f.TaskQueue(taskType).NormalPartition(partitionId),
+		version: PhysicalTaskQueueVersion{
+			versionSet:           versionSet,
+			buildId:              buildId,
+			deploymentSeriesName: deploymentName,
+		},
 	}, nil
 }
 
-func parseSuffix(persistenceName string, suffix string) (partition int, versionSet string, buildId string, err error) {
-	if partitionOff := strings.LastIndex(suffix, buildIdDelimiter); partitionOff == 0 {
-		return 0, "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
+//nolint:revive // cognitive complexity will be simpler once versioning 1-2 is cleaned up
+func parseSuffix(persistenceName string, suffix string) (partition int, versionSet string, buildId string, deploymentName string, err error) {
+	if buildIdOff := strings.LastIndex(suffix, deploymentNameDelimiter); buildIdOff == 0 {
+		return 0, "", "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
+	} else if buildIdOff > 0 {
+		deploymentNameBytes, err := base64.RawURLEncoding.DecodeString(suffix[:buildIdOff])
+		if err != nil {
+			return 0, "", "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
+		}
+		deploymentName = string(deploymentNameBytes)
+
+		suffix = suffix[buildIdOff+1:]
+		if partitionOff := strings.LastIndex(suffix, buildIdDelimiter); partitionOff == 0 {
+			return 0, "", "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
+		} else if partitionOff > 0 {
+			buildIdBytes, err := base64.RawURLEncoding.DecodeString(suffix[:partitionOff])
+			if err != nil {
+				return 0, "", "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
+			}
+			buildId = string(buildIdBytes)
+			suffix = suffix[partitionOff+1:]
+		}
+	} else if partitionOff := strings.LastIndex(suffix, buildIdDelimiter); partitionOff == 0 {
+		return 0, "", "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
 	} else if partitionOff > 0 {
 		buildIdBytes, err := base64.URLEncoding.DecodeString(suffix[:partitionOff])
 		if err != nil {
-			return 0, "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
+			return 0, "", "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
 		}
 		buildId = string(buildIdBytes)
 		suffix = suffix[partitionOff+1:]
 	} else if partitionOff := strings.LastIndex(suffix, versionSetDelimiter); partitionOff == 0 {
-		return 0, "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
+		return 0, "", "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
 	} else if partitionOff > 0 {
 		// pull out version set
 		versionSet, suffix = suffix[:partitionOff], suffix[partitionOff+1:]
 	}
 
 	partition, err = strconv.Atoi(suffix)
 	if err != nil || partition < 0 || (partition == 0 && len(versionSet) == 0 && len(buildId) == 0) {
-		return 0, "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
+		return 0, "", "", "", fmt.Errorf("%w: %s", ErrInvalidPersistenceName, persistenceName)
 	}
-	return partition, versionSet, buildId, err
+	return partition, versionSet, buildId, deploymentName, err
 }
 
 func (q *PhysicalTaskQueueKey) IsVersioned() bool {
-	return q.versionSet != "" || q.buildId != ""
+	return q.version.IsVersioned()
+}
+
+// Version returns a pointer to the physical queue version key. Caller must not manipulate the
+// returned value.
+func (q *PhysicalTaskQueueKey) Version() PhysicalTaskQueueVersion {
+	return q.version
+}
+
+func (v PhysicalTaskQueueVersion) IsVersioned() bool {
+	return v.versionSet != "" || v.buildId != ""
+}
+
+func (v PhysicalTaskQueueVersion) Deployment() *deployment.Deployment {
+	if len(v.deploymentSeriesName) > 0 {
+		return &deployment.Deployment{
+			SeriesName: v.deploymentSeriesName,
+			BuildId:    v.buildId,
+		}
+	}
+	return nil
+}
+
+// BuildId returns empty if this is not a Versioning v2 queue.
+func (v PhysicalTaskQueueVersion) BuildId() string {
+	if len(v.deploymentSeriesName) > 0 {
+		return ""
+	}
+	return v.buildId
+}
+
+func (v PhysicalTaskQueueVersion) VersionSet() string {
+	return v.versionSet
+}
+
+// MetricsTagValue returns the build ID tag value for this version.
+func (v PhysicalTaskQueueVersion) MetricsTagValue() string {
+	if v.versionSet != "" {
+		return v.versionSet
+	} else if v.deploymentSeriesName == "" {
+		return v.buildId
+	}
+	return v.deploymentSeriesName + "/" + v.buildId
 }