Add Mutable State X-Cluster RecordChildCompletion

common/authorization/oauthAutorizer_test.go

@@ -44,14 +44,14 @@ import (
 type (
 	oauthSuite struct {
 		suite.Suite
-		logger          *log.MockLogger
-		cfg             config.OAuthAuthorizer
-		att             Attributes
-		token           string
-		controller      *gomock.Controller
-		domainCache     *cache.MockDomainCache
-		ctx             context.Context
-		domainEntry     *cache.DomainCacheEntry
+		logger      *log.MockLogger
+		cfg         config.OAuthAuthorizer
+		att         Attributes
+		token       string
+		controller  *gomock.Controller
+		domainCache *cache.MockDomainCache
+		ctx         context.Context
+		domainEntry *cache.DomainCacheEntry
 	}
 )
 
@@ -88,12 +88,11 @@ func (s *oauthSuite) SetupTest() {
 
 	s.domainEntry = cache.NewGlobalDomainCacheEntryForTest(
 		&persistence.DomainInfo{
-			ID: "test-domain-id",
+			ID:   "test-domain-id",
 			Name: "test-domain",
 			Data: map[string]string{
 				common.DomainDataKeyForReadGroups: "c",
 			},
-
 		},
 		&persistence.DomainConfig{Retention: 1},
 		&persistence.DomainReplicationConfig{
@@ -227,4 +226,4 @@ func (s *oauthSuite) TestIncorrectPermission() {
 	}))
 	result, _ := authorizer.Authorize(s.ctx, &s.att)
 	s.Equal(result.Decision, DecisionDeny)
-}
+}

common/constants.go

@@ -202,4 +202,4 @@ const StickyTaskConditionFailedErrorMsg = "StickyTaskConditionFailedError"
 const MemoKeyForOperator = "operator"
 
 // ReservedTaskListPrefix is the required naming prefix for any task list partition other than partition 0
-const ReservedTaskListPrefix = "/__cadence_sys/"
+const ReservedTaskListPrefix = "/__cadence_sys/"

common/persistence/dataManagerInterfaces.go

@@ -150,13 +150,15 @@ const (
 	TransferTaskTypeRecordWorkflowStarted
 	TransferTaskTypeResetWorkflow
 	TransferTaskTypeUpsertWorkflowSearchAttributes
+	TransferTaskTypeRecordChildWorkflowExecutionComplete
 )
 
 // Types of cross-cluster tasks
 const (
 	CrossClusterTaskTypeStartChildExecution = iota + 1
 	CrossClusterTaskTypeCancelExecution
 	CrossClusterTaskTypeSignalExecution
+	CrossClusterTaskTypeRecordChildWorkflowExeuctionComplete
 )
 
 // Types of replication tasks
@@ -566,6 +568,17 @@ type (
 		Version             int64
 	}
 
+	// RecordChildCompletionTask identifies a task completing a child execution
+	RecordWorkflowExecutionCompleteTask struct {
+		VisibilityTimestamp time.Time
+		TaskID              int64
+		TargetDomainID      string
+		TargetWorkflowID    string
+		TargetRunID         string
+		InitiatedID         int64
+		Version             int64
+	}
+
 	// CrossClusterStartChildExecutionTask is the cross-cluster version of StartChildExecutionTask
 	CrossClusterStartChildExecutionTask struct {
 		StartChildExecutionTask
@@ -587,6 +600,13 @@ type (
 		TargetCluster string
 	}
 
+	// CrossClusterRecordChildWorkflowExecutionCompleteTask is the cross-cluster version of RecordChildCompletionTask
+	CrossClusterRecordChildWorkflowExecutionCompleteTask struct {
+		RecordWorkflowExecutionCompleteTask
+
+		TargetCluster string
+	}
+
 	// ActivityTimeoutTask identifies a timeout task.
 	ActivityTimeoutTask struct {
 		VisibilityTimestamp time.Time
@@ -2246,6 +2266,41 @@ func (u *SignalExecutionTask) SetVisibilityTimestamp(timestamp time.Time) {
 	u.VisibilityTimestamp = timestamp
 }
 
+// GetType returns the type of the signal transfer task
+func (u *RecordWorkflowExecutionCompleteTask) GetType() int {
+	return TransferTaskTypeRecordChildWorkflowExecutionComplete
+}
+
+// GetVersion returns the version of the signal transfer task
+func (u *RecordWorkflowExecutionCompleteTask) GetVersion() int64 {
+	return u.Version
+}
+
+// SetVersion returns the version of the signal transfer task
+func (u *RecordWorkflowExecutionCompleteTask) SetVersion(version int64) {
+	u.Version = version
+}
+
+// GetTaskID returns the sequence ID of the signal transfer task.
+func (u *RecordWorkflowExecutionCompleteTask) GetTaskID() int64 {
+	return u.TaskID
+}
+
+// SetTaskID sets the sequence ID of the signal transfer task.
+func (u *RecordWorkflowExecutionCompleteTask) SetTaskID(id int64) {
+	u.TaskID = id
+}
+
+// GetVisibilityTimestamp get the visibility timestamp
+func (u *RecordWorkflowExecutionCompleteTask) GetVisibilityTimestamp() time.Time {
+	return u.VisibilityTimestamp
+}
+
+// SetVisibilityTimestamp set the visibility timestamp
+func (u *RecordWorkflowExecutionCompleteTask) SetVisibilityTimestamp(timestamp time.Time) {
+	u.VisibilityTimestamp = timestamp
+}
+
 // GetType returns the type of the upsert search attributes transfer task
 func (u *UpsertWorkflowSearchAttributesTask) GetType() int {
 	return TransferTaskTypeUpsertWorkflowSearchAttributes
@@ -2331,6 +2386,11 @@ func (c *CrossClusterSignalExecutionTask) GetType() int {
 	return CrossClusterTaskTypeSignalExecution
 }
 
+// GetType returns of type of the cross-cluster record child workflow completion task
+func (c *CrossClusterRecordChildWorkflowExecutionCompleteTask) GetType() int {
+	return CrossClusterTaskTypeRecordChildWorkflowExeuctionComplete
+}
+
 // GetType returns the type of the history replication task
 func (a *HistoryReplicationTask) GetType() int {
 	return ReplicationTaskTypeHistory

common/persistence/nosql/nosqlExecutionStoreUtil.go

@@ -348,6 +348,13 @@ func (d *nosqlExecutionStore) prepareCrossClusterTasksForWorkflowTxn(
 			targetChildWorkflowOnly = task.(*p.CrossClusterSignalExecutionTask).TargetChildWorkflowOnly
 			scheduleID = task.(*p.CrossClusterSignalExecutionTask).InitiatedID
 
+		case p.CrossClusterTaskTypeRecordChildWorkflowExeuctionComplete:
+			targetCluster = task.(*p.CrossClusterRecordChildWorkflowExecutionCompleteTask).TargetCluster
+			targetDomainID = task.(*p.CrossClusterRecordChildWorkflowExecutionCompleteTask).TargetDomainID
+			targetWorkflowID = task.(*p.CrossClusterRecordChildWorkflowExecutionCompleteTask).TargetWorkflowID
+			targetRunID = task.(*p.CrossClusterRecordChildWorkflowExecutionCompleteTask).TargetRunID
+			scheduleID = task.(*p.CrossClusterRecordChildWorkflowExecutionCompleteTask).InitiatedID
+
 		default:
 			return nil, &types.InternalServiceError{
 				Message: fmt.Sprintf("Unknown cross-cluster task type: %v", task.GetType()),

common/persistence/sql/sqlExecutionStoreUtil.go

@@ -805,6 +805,13 @@ func createCrossClusterTasks(
 			info.TargetChildWorkflowOnly = task.(*p.CrossClusterSignalExecutionTask).TargetChildWorkflowOnly
 			info.ScheduleID = task.(*p.CrossClusterSignalExecutionTask).InitiatedID
 
+		case p.CrossClusterTaskTypeRecordChildWorkflowExeuctionComplete:
+			crossClusterTasksRows[i].TargetCluster = task.(*p.CrossClusterRecordChildWorkflowExecutionCompleteTask).TargetCluster
+			info.TargetDomainID = serialization.MustParseUUID(task.(*p.CrossClusterRecordChildWorkflowExecutionCompleteTask).TargetDomainID)
+			info.TargetWorkflowID = task.(*p.CrossClusterRecordChildWorkflowExecutionCompleteTask).TargetWorkflowID
+			info.TargetRunID = serialization.MustParseUUID(task.(*p.CrossClusterRecordChildWorkflowExecutionCompleteTask).TargetRunID)
+			info.ScheduleID = task.(*p.CrossClusterRecordChildWorkflowExecutionCompleteTask).InitiatedID
+
 		default:
 			return &types.InternalServiceError{
 				Message: fmt.Sprintf("Unknown cross-cluster task type: %v", task.GetType()),

common/types/mapper/proto/shared.go

@@ -1360,6 +1360,47 @@ func ToCrossClusterSignalExecutionResponseAttributes(t *sharedv1.CrossClusterSig
 	return &types.CrossClusterSignalExecutionResponseAttributes{}
 }
 
+// FromCrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes converts internal CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes type to proto
+func FromCrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes(t *types.CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes) *sharedv1.CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes {
+	if t == nil {
+		return nil
+	}
+	return &sharedv1.CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes{
+		TargetDomainId:          t.TargetDomainID,
+		TargetWorkflowExecution: FromWorkflowRunPair(t.TargetWorkflowID, t.TargetRunID),
+		InitiatedEventId:        t.InitiatedEventID,
+	}
+}
+
+// ToCrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes converts proto CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes type to internal
+func ToCrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes(t *sharedv1.CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes) *types.CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes {
+	if t == nil {
+		return nil
+	}
+	return &types.CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes{
+		TargetDomainID:   t.TargetDomainId,
+		TargetWorkflowID: ToWorkflowID(t.TargetWorkflowExecution),
+		TargetRunID:      ToRunID(t.TargetWorkflowExecution),
+		InitiatedEventID: t.InitiatedEventId,
+	}
+}
+
+// FromCrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes converts internal CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes type to proto
+func FromCrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes(t *types.CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes) *sharedv1.CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes {
+	if t == nil {
+		return nil
+	}
+	return &sharedv1.CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes{}
+}
+
+// ToCrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes converts proto CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes type to internal
+func ToCrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes(t *sharedv1.CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes) *types.CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes {
+	if t == nil {
+		return nil
+	}
+	return &types.CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes{}
+}
+
 // FromCrossClusterTaskRequest converts internal CrossClusterTaskRequest type to proto
 func FromCrossClusterTaskRequest(t *types.CrossClusterTaskRequest) *sharedv1.CrossClusterTaskRequest {
 	if t == nil {

common/types/shared.go

@@ -10652,12 +10652,56 @@ func (v *CrossClusterSignalExecutionRequestAttributes) GetControl() (o []byte) {
 type CrossClusterSignalExecutionResponseAttributes struct {
 }
 
+type CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes struct {
+	TargetDomainID   string `json:"targetDomainID,omitempty"`
+	TargetWorkflowID string `json:"targetWorkflowID,omitempty"`
+	TargetRunID      string `json:"targetRunID,omitempty"`
+	InitiatedEventID int64  `json:"initiatedEventID,omitempty"`
+}
+
+// GetTargetDomainID is an internal getter (TBD...)
+func (v *CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes) GetTargetDomainID() (o string) {
+	if v != nil {
+		return v.TargetDomainID
+	}
+	return
+}
+
+// GetTargetWorkflowID is an internal getter (TBD...)
+func (v *CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes) GetTargetWorkflowID() (o string) {
+	if v != nil {
+		return v.TargetWorkflowID
+	}
+	return
+}
+
+// GetTargetRunID is an internal getter (TBD...)
+func (v *CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes) GetTargetRunID() (o string) {
+	if v != nil {
+		return v.TargetRunID
+	}
+	return
+}
+
+// GetInitiatedEventID is an internal getter (TBD...)
+func (v *CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes) GetInitiatedEventID() (o int64) {
+	if v != nil {
+		return v.InitiatedEventID
+	}
+	return
+}
+
+// CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes is an internal type (TBD...)
+type CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes struct {
+}
+
 // CrossClusterTaskRequest is an internal type (TBD...)
 type CrossClusterTaskRequest struct {
-	TaskInfo                      *CrossClusterTaskInfo                             `json:"taskInfo,omitempty"`
-	StartChildExecutionAttributes *CrossClusterStartChildExecutionRequestAttributes `json:"startChildExecutionAttributes,omitempty"`
-	CancelExecutionAttributes     *CrossClusterCancelExecutionRequestAttributes     `json:"cancelExecutionAttributes,omitempty"`
-	SignalExecutionAttributes     *CrossClusterSignalExecutionRequestAttributes     `json:"signalExecutionAttributes,omitempty"`
+	TaskInfo                                       *CrossClusterTaskInfo                                              `json:"taskInfo,omitempty"`
+	StartChildExecutionAttributes                  *CrossClusterStartChildExecutionRequestAttributes                  `json:"startChildExecutionAttributes,omitempty"`
+	CancelExecutionAttributes                      *CrossClusterCancelExecutionRequestAttributes                      `json:"cancelExecutionAttributes,omitempty"`
+	SignalExecutionAttributes                      *CrossClusterSignalExecutionRequestAttributes                      `json:"signalExecutionAttributes,omitempty"`
+	RecordChildWorkflowExecutionCompleteAttributes *CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes `json:"RecordChildWorkflowExecutionCompleteAttributes,omitempty"`
 }
 
 // GetTaskInfo is an internal getter (TBD...)
@@ -10692,14 +10736,23 @@ func (v *CrossClusterTaskRequest) GetSignalExecutionAttributes() (o *CrossCluste
 	return
 }
 
+// GetSignalExecutionAttributes is an internal getter (TBD...)
+func (v *CrossClusterTaskRequest) GetRecordChildWorkflowExecutionCompleteAttributes() (o *CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes) {
+	if v != nil && v.RecordChildWorkflowExecutionCompleteAttributes != nil {
+		return v.RecordChildWorkflowExecutionCompleteAttributes
+	}
+	return
+}
+
 // CrossClusterTaskResponse is an internal type (TBD...)
 type CrossClusterTaskResponse struct {
-	TaskID                        int64                                              `json:"taskID,omitempty"`
-	TaskType                      *CrossClusterTaskType                              `json:"taskType,omitempty"`
-	FailedCause                   *CrossClusterTaskFailedCause                       `json:"failedCause,omitempty"`
-	StartChildExecutionAttributes *CrossClusterStartChildExecutionResponseAttributes `json:"startChildExecutionAttributes,omitempty"`
-	CancelExecutionAttributes     *CrossClusterCancelExecutionResponseAttributes     `json:"cancelExecutionAttributes,omitempty"`
-	SignalExecutionAttributes     *CrossClusterSignalExecutionResponseAttributes     `json:"signalExecutionAttributes,omitempty"`
+	TaskID                                         int64                                                               `json:"taskID,omitempty"`
+	TaskType                                       *CrossClusterTaskType                                               `json:"taskType,omitempty"`
+	FailedCause                                    *CrossClusterTaskFailedCause                                        `json:"failedCause,omitempty"`
+	StartChildExecutionAttributes                  *CrossClusterStartChildExecutionResponseAttributes                  `json:"startChildExecutionAttributes,omitempty"`
+	CancelExecutionAttributes                      *CrossClusterCancelExecutionResponseAttributes                      `json:"cancelExecutionAttributes,omitempty"`
+	SignalExecutionAttributes                      *CrossClusterSignalExecutionResponseAttributes                      `json:"signalExecutionAttributes,omitempty"`
+	RecordChildWorkflowExecutionCompleteAttributes *CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes `json:"RecordChildWorkflowExecutionCompleteAttributes,omitempty"`
 }
 
 // GetTaskID is an internal getter (TBD...)
@@ -10750,6 +10803,14 @@ func (v *CrossClusterTaskResponse) GetSignalExecutionAttributes() (o *CrossClust
 	return
 }
 
+// GetSignalExecutionAttributes is an internal getter (TBD...)
+func (v *CrossClusterTaskResponse) GetRecordChildWorkflowExecutionCompleteAttributes() (o *CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes) {
+	if v != nil && v.RecordChildWorkflowExecutionCompleteAttributes != nil {
+		return v.RecordChildWorkflowExecutionCompleteAttributes
+	}
+	return
+}
+
 // GetCrossClusterTasksRequest is an internal type (TBD...)
 type GetCrossClusterTasksRequest struct {
 	ShardIDs      []int32 `json:"shardIDs,omitempty"`

proto/internal/uber/cadence/shared/v1/queue.proto

@@ -75,7 +75,7 @@ message CrossClusterStartChildExecutionRequestAttributes {
   string request_id = 2;
   int64 initiated_event_id = 3;
   api.v1.StartChildWorkflowExecutionInitiatedEventAttributes initiated_event_attributes = 4;
-  // targetRunID is for scheduling first decision task 
+  // targetRunID is for scheduling first decision task
   // targetWorkflowID is available in initiatedEventAttributes
   string target_run_id = 5;
 }
@@ -109,12 +109,22 @@ message CrossClusterSignalExecutionRequestAttributes {
 message CrossClusterSignalExecutionResponseAttributes {
 }
 
+message CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes {
+  string target_domain_id = 1;
+  api.v1.WorkflowExecution target_workflow_execution = 2;
+  int64 initiated_event_id = 3;
+}
+
+message CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes {
+}
+
 message CrossClusterTaskRequest {
   CrossClusterTaskInfo task_info = 1;
   oneof attributes {
     CrossClusterStartChildExecutionRequestAttributes startChildExecutionAttributes = 2;
     CrossClusterCancelExecutionRequestAttributes cancelExecutionAttributes = 3;
     CrossClusterSignalExecutionRequestAttributes signalExecutionAttributes = 4;
+    CrossClusterRecordChildWorkflowExecutionCompleteRequestAttributes recordChildWorkflowExecutionCompleteRequestAttributes = 5;
   }
 }
 
@@ -126,6 +136,7 @@ message CrossClusterTaskResponse {
     CrossClusterStartChildExecutionResponseAttributes startChildExecutionAttributes = 4;
     CrossClusterCancelExecutionResponseAttributes cancelExecutionAttributes = 5;
     CrossClusterSignalExecutionResponseAttributes signalExecutionAttributes = 6;
+    CrossClusterRecordChildWorkflowExecutionCompleteResponseAttributes recordChildWorkflowExecutionCompleteRequestAttributes = 7;
   }
 }
 

service/history/execution/mutable_state_task_generator.go

@@ -600,7 +600,18 @@ func (r *mutableStateTaskGeneratorImpl) GenerateCrossClusterTaskFromTransferTask
 				Version:          task.Version,
 			},
 		}
-	// TODO: add the case for TransferTaskTypeCloseExecution
+	case persistence.TransferTaskTypeRecordChildWorkflowExecutionComplete:
+		crossClusterTask = &persistence.CrossClusterRecordChildWorkflowExecutionCompleteTask{
+			TargetCluster: targetCluster,
+			RecordWorkflowExecutionCompleteTask: persistence.RecordWorkflowExecutionCompleteTask{
+				// TaskID is set by shard context
+				TargetDomainID:   task.TargetDomainID,
+				TargetWorkflowID: task.TargetWorkflowID,
+				TargetRunID:      task.TargetRunID,
+				InitiatedID:      task.ScheduleID,
+				Version:          task.Version,
+			},
+		}
 	default:
 		return fmt.Errorf("unable to convert transfer task of type %v to cross-cluster task", task.TaskType)
 	}