Define type and name for task categories (#2522)

service/history/shard/context_impl.go

@@ -1058,8 +1058,8 @@ func (s *ContextImpl) allocateTaskIDsLocked(
 	transferMaxReadLevel *int64,
 ) error {
 	currentCluster := s.GetClusterMetadata().GetCurrentClusterName()
-	for _, tasks := range newTasks {
-		for _, task := range tasks {
+	for category, tasksByCategory := range newTasks {
+		for _, task := range tasksByCategory {
 			// set taskID
 			id, err := s.generateTransferTaskIDLocked()
 			if err != nil {
@@ -1070,7 +1070,7 @@ func (s *ContextImpl) allocateTaskIDsLocked(
 			*transferMaxReadLevel = id
 
 			// if scheduled task, check if fire time is in the past
-			if !task.GetKey().FireTime.IsZero() {
+			if category.Type() == tasks.CategoryTypeScheduled {
 				ts := task.GetVisibilityTime()
 				if task.GetVersion() != common.EmptyVersion {
 					// cannot use version to determine the corresponding cluster for timer task

service/history/tasks/category.go

@@ -0,0 +1,141 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package tasks
+
+import (
+	"fmt"
+	"sync"
+)
+
+type (
+	Category struct {
+		id    int32
+		cType CategoryType
+		name  string
+	}
+
+	CategoryType int
+)
+
+const (
+	CategoryIDUnspecified int32 = iota
+	CategoryIDTransfer
+	CategoryIDTimer
+	CategoryIDVisibility
+	CategoryIDReplication
+)
+
+const (
+	CategoryTypeUnspecified CategoryType = iota
+	CategoryTypeImmediate
+	CategoryTypeScheduled
+)
+
+const (
+	CategoryNameTransfer    = "transfer"
+	CategoryNameTimer       = "timer"
+	CategoryNameVisibility  = "visibility"
+	CategoryNameReplication = "replication"
+)
+
+var (
+	CategoryTransfer = Category{
+		id:    CategoryIDTransfer,
+		cType: CategoryTypeImmediate,
+		name:  CategoryNameTransfer,
+	}
+
+	CategoryTimer = Category{
+		id:    CategoryIDTimer,
+		cType: CategoryTypeScheduled,
+		name:  CategoryNameTimer,
+	}
+
+	CategoryVisibility = Category{
+		id:    CategoryIDVisibility,
+		cType: CategoryTypeImmediate,
+		name:  CategoryNameVisibility,
+	}
+
+	CategoryReplication = Category{
+		id:    CategoryIDReplication,
+		cType: CategoryTypeImmediate,
+		name:  CategoryNameReplication,
+	}
+)
+
+var (
+	categories = struct {
+		sync.Mutex
+		list []Category
+	}{
+		list: []Category{
+			CategoryTransfer,
+			CategoryTimer,
+			CategoryVisibility,
+			CategoryReplication,
+		},
+	}
+)
+
+func NewCategory(
+	id int32,
+	categoryType CategoryType,
+	name string,
+) Category {
+	categories.Lock()
+	defer categories.Unlock()
+
+	for _, existingCategory := range categories.list {
+		if existingCategory.ID() == id {
+			panic(fmt.Sprintf("category id: %v has already been used", id))
+		}
+
+		if existingCategory.Name() == name {
+			panic(fmt.Sprintf("categeory name: %s has already been used", name))
+		}
+	}
+
+	newCategory := Category{
+		id:    id,
+		cType: categoryType,
+		name:  name,
+	}
+	categories.list = append(categories.list, newCategory)
+
+	return newCategory
+}
+
+func (c *Category) ID() int32 {
+	return c.id
+}
+
+func (c *Category) Name() string {
+	return c.name
+}
+
+func (c *Category) Type() CategoryType {
+	return c.cType
+}

service/history/tasks/task.go

@@ -38,8 +38,6 @@ type (
 
 	Keys []Key
 
-	Category int32
-
 	// Task is the generic task interface
 	Task interface {
 		GetKey() Key
@@ -57,14 +55,6 @@ type (
 	}
 )
 
-const (
-	CategoryUnspecified Category = iota
-	CategoryTransfer
-	CategoryTimer
-	CategoryVisibility
-	CategoryReplication
-)
-
 func (left Key) CompareTo(right Key) int {
 	if left.FireTime.Before(right.FireTime) {
 		return -1

service/history/workflow/mutable_state_util.go

@@ -73,7 +73,7 @@ func setTaskInfo(
 
 		for _, task := range tasksByCategory {
 			task.SetVersion(version)
-			if task.GetKey().FireTime.IsZero() {
+			if category.Type() == tasks.CategoryTypeImmediate {
 				task.SetVisibilityTime(timestamp)
 			}
 		}