Transfer Queue Processor can get stuck on StartChildExecution task (#337)

Handle EntityNotExistError in the processing of transfer task for
StartChildExecution.

Author: samarabbas

common/persistence/persistenceTestBase.go

@@ -519,6 +519,18 @@ func (s *TestBase) UpdateWorkflowExecutionWithTransferTasks(
 	})
 }
 
+// UpdateWorkflowExecutionForChildExecutionsInitiated is a utility method to update workflow execution
+func (s *TestBase) UpdateWorkflowExecutionForChildExecutionsInitiated(
+	updatedInfo *WorkflowExecutionInfo, condition int64, transferTasks []Task, childInfos []*ChildExecutionInfo) error {
+	return s.WorkflowMgr.UpdateWorkflowExecution(&UpdateWorkflowExecutionRequest{
+		ExecutionInfo:             updatedInfo,
+		TransferTasks:             transferTasks,
+		Condition:                 condition,
+		UpsertChildExecutionInfos: childInfos,
+		RangeID:                   s.ShardInfo.RangeID,
+	})
+}
+
 // UpdateWorkflowExecutionForRequestCancel is a utility method to update workflow execution
 func (s *TestBase) UpdateWorkflowExecutionForRequestCancel(
 	updatedInfo *WorkflowExecutionInfo, condition int64, transferTasks []Task,

service/history/historyEngine_test.go

@@ -2512,6 +2512,24 @@ func addRequestCancelInitiatedEvent(builder *mutableStateBuilder, decisionComple
 	return event
 }
 
+func addStartChildWorkflowExecutionInitiatedEvent(builder *mutableStateBuilder, decisionCompletedID int64,
+	createRequestID, domain, workflowID, workflowType, tasklist string, input []byte,
+	executionStartToCloseTimeout, taskStartToCloseTimeout int32) (*workflow.HistoryEvent,
+	*persistence.ChildExecutionInfo) {
+	return builder.AddStartChildWorkflowExecutionInitiatedEvent(decisionCompletedID, createRequestID,
+		&workflow.StartChildWorkflowExecutionDecisionAttributes{
+			Domain:       common.StringPtr(domain),
+			WorkflowId:   common.StringPtr(workflowID),
+			WorkflowType: &workflow.WorkflowType{Name: common.StringPtr(workflowType)},
+			TaskList:     &workflow.TaskList{Name: common.StringPtr(tasklist)},
+			Input:        input,
+			ExecutionStartToCloseTimeoutSeconds: common.Int32Ptr(executionStartToCloseTimeout),
+			TaskStartToCloseTimeoutSeconds:      common.Int32Ptr(taskStartToCloseTimeout),
+			ChildPolicy:                         workflow.ChildPolicyPtr(workflow.ChildPolicy_TERMINATE),
+			Control:                             nil,
+		})
+}
+
 func addCompleteWorkflowEvent(builder *mutableStateBuilder, decisionCompletedEventID int64,
 	result []byte) *workflow.HistoryEvent {
 	e := builder.AddCompletedWorkflowEvent(decisionCompletedEventID, &workflow.CompleteWorkflowExecutionDecisionAttributes{

service/history/transferQueueProcessor.go

@@ -651,21 +651,15 @@ func (t *transferQueueProcessorImpl) processStartChildExecution(task *persistenc
 				return err
 			}
 			// Finally create first decision task for Child execution so it is really started
-			err = t.historyClient.ScheduleDecisionTask(nil, &history.ScheduleDecisionTaskRequest{
-				DomainUUID: common.StringPtr(targetDomainID),
-				WorkflowExecution: &workflow.WorkflowExecution{
-					WorkflowId: common.StringPtr(task.TargetWorkflowID),
-					RunId:      common.StringPtr(startResponse.GetRunId()),
-				},
+			err = t.createFirstDecisionTask(targetDomainID, &workflow.WorkflowExecution{
+				WorkflowId: common.StringPtr(task.TargetWorkflowID),
+				RunId:      common.StringPtr(startResponse.GetRunId()),
 			})
 		} else {
 			// ChildExecution already started, just create DecisionTask and complete transfer task
 			startedEvent, _ := msBuilder.GetChildExecutionStartedEvent(initiatedEventID)
 			startedAttributes := startedEvent.GetChildWorkflowExecutionStartedEventAttributes()
-			err = t.historyClient.ScheduleDecisionTask(nil, &history.ScheduleDecisionTaskRequest{
-				DomainUUID:        common.StringPtr(targetDomainID),
-				WorkflowExecution: startedAttributes.GetWorkflowExecution(),
-			})
+			err = t.createFirstDecisionTask(targetDomainID, startedAttributes.GetWorkflowExecution())
 		}
 	}
 
@@ -744,6 +738,26 @@ func (t *transferQueueProcessorImpl) recordStartChildExecutionFailed(task *persi
 		})
 }
 
+// createFirstDecisionTask is used by StartChildExecution transfer task to create the first decision task for
+// child execution.
+func (t *transferQueueProcessorImpl) createFirstDecisionTask(domainID string,
+	execution *workflow.WorkflowExecution) error {
+	err := t.historyClient.ScheduleDecisionTask(nil, &history.ScheduleDecisionTaskRequest{
+		DomainUUID:        common.StringPtr(domainID),
+		WorkflowExecution: execution,
+	})
+
+	if err != nil {
+		if _, ok := err.(*workflow.EntityNotExistsError); ok {
+			// Maybe child workflow execution already timedout or terminated
+			// Safe to discard the error and complete this transfer task
+			return nil
+		}
+	}
+
+	return err
+}
+
 func (t *transferQueueProcessorImpl) requestCancelCompleted(task *persistence.TransferTaskInfo,
 	context *workflowExecutionContext,
 	request *history.RequestCancelWorkflowExecutionRequest) error {

service/history/transferQueueProcessor_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
 
+	"github.com/pborman/uuid"
 	"github.com/uber-common/bark"
 	m "github.com/uber/cadence/.gen/go/matching"
 	workflow "github.com/uber/cadence/.gen/go/shared"
@@ -424,6 +425,108 @@ workerPump:
 	}
 }
 
+func (s *transferQueueProcessorSuite) TestStartChildExecutionTransferTasks() {
+	domain := "start-child-execution-transfer-tasks-test-domain"
+	domainID := "b7f71853-0a8c-4eb1-af6b-c4e71dae50a1"
+	workflowID := "start-child-execution-transfertasks-test"
+	runID := "67ad6d62-79c6-4c8a-a27a-1fb7a10ae789"
+	workflowExecution := workflow.WorkflowExecution{
+		WorkflowId: common.StringPtr(workflowID),
+		RunId:      common.StringPtr(runID),
+	}
+	taskList := "start-child-execution-transfertasks-queue"
+	identity := "start-child-execution-transfertasks-test"
+
+	tasksCh := s.createChildExecutionState(domain, domainID, workflowExecution, taskList, identity)
+	childRunID := "d3f9164e-b696-4350-8409-9a6cf670f4f2"
+workerPump:
+	for {
+		select {
+		case task := <-tasksCh:
+			if task.TaskType == persistence.TransferTaskTypeStartChildExecution {
+				s.mockHistoryClient.On("StartWorkflowExecution", mock.Anything, mock.Anything).Once().Return(
+					&workflow.StartWorkflowExecutionResponse{
+						RunId: common.StringPtr(childRunID),
+					}, nil)
+				s.mockHistoryClient.On("ScheduleDecisionTask", mock.Anything, mock.Anything).Once().Return(nil)
+			}
+			s.processor.processTransferTask(task)
+		default:
+			break workerPump
+		}
+	}
+}
+
+func (s *transferQueueProcessorSuite) TestStartChildExecutionTransferTasksChildCompleted() {
+	domain := "start-child-execution-transfer-tasks-child-completed-test-domain"
+	domainID := "8bf7aaf8-810a-4778-a549-d7913d2a5b82"
+	workflowID := "start-child-execution-transfertasks-child-completed-test"
+	runID := "a627ea38-7e5e-41cc-9a32-4c7979b93ae2"
+	workflowExecution := workflow.WorkflowExecution{
+		WorkflowId: common.StringPtr(workflowID),
+		RunId:      common.StringPtr(runID),
+	}
+	taskList := "start-child-execution-transfertasks-child-completed-queue"
+	identity := "start-child-execution-transfertasks-child-completed-test"
+
+	tasksCh := s.createChildExecutionState(domain, domainID, workflowExecution, taskList, identity)
+	childRunID := "66825b60-5ae2-4ff3-8da6-cbb286b4a7e6"
+workerPump:
+	for {
+		select {
+		case task := <-tasksCh:
+			if task.TaskType == persistence.TransferTaskTypeStartChildExecution {
+				s.mockHistoryClient.On("StartWorkflowExecution", mock.Anything, mock.Anything).Once().Return(
+					&workflow.StartWorkflowExecutionResponse{
+						RunId: common.StringPtr(childRunID),
+					}, nil)
+				s.mockHistoryClient.On("ScheduleDecisionTask", mock.Anything, mock.Anything).Once().Return(
+					&workflow.EntityNotExistsError{})
+			}
+			s.processor.processTransferTask(task)
+		default:
+			break workerPump
+		}
+	}
+}
+
+func (s *transferQueueProcessorSuite) createChildExecutionState(domain, domainID string,
+	workflowExecution workflow.WorkflowExecution, taskList, identity string) chan *persistence.TransferTaskInfo {
+	_, err0 := s.CreateWorkflowExecution(domainID, workflowExecution, taskList, "wType", 10, nil, 3, 0, 2, nil)
+	s.Nil(err0, "No error expected.")
+	s.mockMatching.On("AddDecisionTask", mock.Anything, mock.Anything).Once().Return(nil)
+	s.mockVisibilityMgr.On("RecordWorkflowExecutionStarted", mock.Anything).Once().Return(nil)
+
+	builder := newMutableStateBuilder(s.ShardContext.GetConfig(), s.logger)
+	info1, _ := s.GetWorkflowExecutionInfo(domainID, workflowExecution)
+	builder.Load(info1)
+	startedEvent := addDecisionTaskStartedEvent(builder, int64(2), taskList, identity)
+	completedEvent := addDecisionTaskCompletedEvent(builder, int64(2), startedEvent.GetEventId(), nil, identity)
+
+	transferTasks := []persistence.Task{}
+	createRequestID := uuid.New()
+
+	childWorkflowID := "start-child-execution-transfertasks-test-child-workflow-id"
+	childWorkflowType := "child-workflow-type"
+	_, ci := addStartChildWorkflowExecutionInitiatedEvent(builder, completedEvent.GetEventId(), createRequestID,
+		domain, childWorkflowID, childWorkflowType, taskList, nil, int32(100), int32(10))
+	transferTasks = append(transferTasks, &persistence.StartChildExecutionTask{
+		TargetDomainID:   domainID,
+		TargetWorkflowID: childWorkflowID,
+		InitiatedID:      ci.InitiatedID,
+	})
+
+	updatedInfo := copyWorkflowExecutionInfo(builder.executionInfo)
+	err1 := s.UpdateWorkflowExecutionForChildExecutionsInitiated(updatedInfo, int64(3), transferTasks,
+		builder.updateChildExecutionInfos)
+	s.Nil(err1, "No error expected.")
+
+	tasksCh := make(chan *persistence.TransferTaskInfo, 10)
+	s.processor.processTransferTasks(tasksCh)
+
+	return tasksCh
+}
+
 func createAddRequestFromTask(task *persistence.TransferTaskInfo, scheduleToStartTimeout int32) interface{} {
 	var res interface{}
 	domainID := task.DomainID