Restrict clustering key in taskList rangeID check (#292)

The CreateTasks query used to write task list tasks to persistence includes a CAS check that verifies that the rangeID has not changed.
The rangeID is a static field in the tasks table, and Cassandra only accepts the partition keys (but not the other clustering keys) in the CAS query to read and update it.
This unrestricted query however causes problems when there is a big backlog in the task list as the coordinator automatically turns it into a LIMIT 1 query.
To fix this we restrict the query with clustering keys, and rewrite non-static fields so that Cassandra accepts the query restrictions.
Issue #277

Author: Tamer Eldeeb

common/persistence/cassandraPersistence.go

@@ -71,8 +71,8 @@ const (
 )
 
 const (
-	taskListTaskID = -12345 // for debugging
-	initialRangeID = 1      // Id of the first range of a new task list
+	taskListTaskID = -12345
+	initialRangeID = 1 // Id of the first range of a new task list
 )
 
 const (
@@ -477,13 +477,6 @@ const (
 		`and type = ? ` +
 		`and task_id = ? ` +
 		`IF range_id = ?`
-
-	templateUpdateTaskListRangeOnlyQuery = `UPDATE tasks SET ` +
-		`range_id = ? ` +
-		`WHERE domain_id = ? ` +
-		`and task_list_name = ? ` +
-		`and task_list_type = ? ` +
-		`IF range_id = ?`
 )
 
 var (
@@ -1196,7 +1189,7 @@ func (d *cassandraPersistence) LeaseTaskList(request *LeaseTaskListRequest) (*Le
 			Msg: fmt.Sprintf("LeaseTaskList failed to apply. db rangeID %v", previousRangeID),
 		}
 	}
-	tli := &TaskListInfo{Name: request.TaskList, TaskType: request.TaskType, RangeID: rangeID + 1, AckLevel: ackLevel}
+	tli := &TaskListInfo{DomainID: request.DomainID, Name: request.TaskList, TaskType: request.TaskType, RangeID: rangeID + 1, AckLevel: ackLevel}
 	return &LeaseTaskListResponse{TaskListInfo: tli}, nil
 }
 
@@ -1244,9 +1237,10 @@ func (d *cassandraPersistence) UpdateTaskList(request *UpdateTaskListRequest) (*
 // From TaskManager interface
 func (d *cassandraPersistence) CreateTasks(request *CreateTasksRequest) (*CreateTasksResponse, error) {
 	batch := d.session.NewBatch(gocql.LoggedBatch)
-	domainID := request.DomainID
-	taskList := request.TaskList
-	taskListType := request.TaskListType
+	domainID := request.TaskListInfo.DomainID
+	taskList := request.TaskListInfo.Name
+	taskListType := request.TaskListInfo.TaskType
+	ackLevel := request.TaskListInfo.AckLevel
 
 	for _, task := range request.Tasks {
 		scheduleID := task.Data.ScheduleID
@@ -1277,13 +1271,20 @@ func (d *cassandraPersistence) CreateTasks(request *CreateTasksRequest) (*Create
 	}
 
 	// The following query is used to ensure that range_id didn't change
-	batch.Query(templateUpdateTaskListRangeOnlyQuery,
-		request.RangeID,
+	batch.Query(templateUpdateTaskListQuery,
+		request.TaskListInfo.RangeID,
 		domainID,
 		taskList,
 		taskListType,
-		request.RangeID,
+		ackLevel,
+		domainID,
+		taskList,
+		taskListType,
+		rowTypeTaskList,
+		taskListTaskID,
+		request.TaskListInfo.RangeID,
 	)
+
 	previous := make(map[string]interface{})
 	applied, _, err := d.session.MapExecuteBatchCAS(batch, previous)
 	if err != nil {
@@ -1295,7 +1296,7 @@ func (d *cassandraPersistence) CreateTasks(request *CreateTasksRequest) (*Create
 		rangeID := previous["range_id"]
 		return nil, &ConditionFailedError{
 			Msg: fmt.Sprintf("Failed to create task. TaskList: %v, taskListType: %v, rangeID: %v, db rangeID: %v",
-				taskList, taskListType, request.RangeID, rangeID),
+				taskList, taskListType, request.TaskListInfo.RangeID, rangeID),
 		}
 	}
 

common/persistence/dataInterfaces.go

@@ -460,10 +460,7 @@ type (
 
 	// CreateTasksRequest is used to create a new task for a workflow exectution
 	CreateTasksRequest struct {
-		DomainID     string
-		TaskList     string
-		TaskListType int
-		RangeID      int64
+		TaskListInfo *TaskListInfo
 		Tasks        []*CreateTaskInfo
 	}
 

common/persistence/persistenceTestBase.go

@@ -723,11 +723,8 @@ func (s *TestBase) CreateDecisionTask(domainID string, workflowExecution workflo
 	}
 
 	_, err = s.TaskMgr.CreateTasks(&CreateTasksRequest{
-		DomainID:     domainID,
-		TaskList:     taskList,
-		TaskListType: TaskListTypeDecision,
+		TaskListInfo: leaseResponse.TaskListInfo,
 		Tasks:        tasks,
-		RangeID:      leaseResponse.TaskListInfo.RangeID,
 	})
 
 	if err != nil {
@@ -766,11 +763,8 @@ func (s *TestBase) CreateActivityTasks(domainID string, workflowExecution workfl
 			},
 		}
 		_, err := s.TaskMgr.CreateTasks(&CreateTasksRequest{
-			DomainID:     domainID,
-			TaskList:     taskList,
-			TaskListType: TaskListTypeActivity,
+			TaskListInfo: leaseResponse.TaskListInfo,
 			Tasks:        tasks,
-			RangeID:      leaseResponse.TaskListInfo.RangeID,
 		})
 
 		if err != nil {

service/matching/matchingEngine_test.go

@@ -1281,28 +1281,29 @@ func (m *testTaskManager) CompleteTask(request *persistence.CompleteTaskRequest)
 
 // CreateTask provides a mock function with given fields: request
 func (m *testTaskManager) CreateTasks(request *persistence.CreateTasksRequest) (*persistence.CreateTasksResponse, error) {
-	domainID := request.DomainID
-	taskList := request.TaskList
-	taskType := request.TaskListType
+	domainID := request.TaskListInfo.DomainID
+	taskList := request.TaskListInfo.Name
+	taskType := request.TaskListInfo.TaskType
+	rangeID := request.TaskListInfo.RangeID
 
 	tlm := m.getTaskListManager(newTaskListID(domainID, taskList, taskType))
 	tlm.Lock()
 	defer tlm.Unlock()
 
 	// First validate the entire batch
 	for _, task := range request.Tasks {
-		m.logger.Debugf("testTaskManager.CreateTask taskID=%v, rangeID=%v", task.TaskID, request.RangeID)
+		m.logger.Debugf("testTaskManager.CreateTask taskID=%v, rangeID=%v", task.TaskID, rangeID)
 		if task.TaskID <= 0 {
 			panic(fmt.Errorf("Invalid taskID=%v", task.TaskID))
 		}
 
-		if tlm.rangeID != request.RangeID {
+		if tlm.rangeID != rangeID {
 			m.logger.Debugf("testTaskManager.CreateTask ConditionFailedError taskID=%v, rangeID: %v, db rangeID: %v",
-				task.TaskID, request.RangeID, tlm.rangeID)
+				task.TaskID, rangeID, tlm.rangeID)
 
 			return nil, &persistence.ConditionFailedError{
 				Msg: fmt.Sprintf("testTaskManager.CreateTask failed. TaskList: %v, taskType: %v, rangeID: %v, db rangeID: %v",
-					taskList, taskType, request.RangeID, tlm.rangeID),
+					taskList, taskType, rangeID, tlm.rangeID),
 			}
 		}
 		_, ok := tlm.tasks.Get(task.TaskID)

service/matching/taskWriter.go

@@ -159,15 +159,20 @@ writerLoop:
 					maxReadLevel = taskIDs[i]
 				}
 
+				tlInfo := &persistence.TaskListInfo{
+					DomainID: w.taskListID.domainID,
+					Name:     w.taskListID.taskListName,
+					TaskType: w.taskListID.taskType,
+					// Note that newTaskID could increment range, so rangeID parameter
+					// might be out of sync. This is OK as caller can just retry.
+					RangeID:  rangeID,
+					AckLevel: w.tlMgr.getAckLevel(),
+				}
+
 				w.tlMgr.persistenceLock.Lock()
 				r, err := w.taskManager.CreateTasks(&persistence.CreateTasksRequest{
-					DomainID:     w.taskListID.domainID,
-					TaskList:     w.taskListID.taskListName,
-					TaskListType: w.taskListID.taskType,
+					TaskListInfo: tlInfo,
 					Tasks:        tasks,
-					// Note that newTaskID could increment range, so rangeID parameter
-					// might be out of sync. This is OK as caller can just retry.
-					RangeID: rangeID,
 				})
 				w.tlMgr.persistenceLock.Unlock()