No-op close transfer task for SyncWorkflowState (temporalio#8507)

michaely520 · web-flow · commit 70dcf172629d · 2025-10-28T11:55:52.000-07:00
## What changed?
No-oping close transfer tasks for SyncWorkflowState tasks


## Why?
We need non state based replication to be eligible for this
optimization.


## How did you test it?
- [x] built
- [x] run locally and tested manually
- [x] covered by existing tests
- [ ] added new unit test(s)
- [x] added new functional test(s)

`go test -v -tags test_dep ./tests/xdc -run
TestStreamBasedReplicationTestSuite/DisableTransitionHistory/TestCloseTransferTaskAckedReplication
-timeout 10m -count=1`

shows

`2025-10-20T08:15:08.290-0700 info Skipping close transfer task
generation - already acked on active cluster {"cluster-name":
"standby_aadnd", "host": "127.0.0.1:57179", "shard-id": 1, "address":
"127.0.0.1:57179", "wf-namespace-id":
"e550305e-0b43-4bcd-a490-8e3223f51ce1", "wf-id":
"test-replication-e2c094d3-c34f-42d9-a166-a967d4e7f602", "wf-run-id":
"019a0230-299a-74ba-a31f-ac247c05f2b9", "logging-call-at":
"/Users/michaely520/projects/temporal/service/history/workflow/task_generator.go:206"}
stream_based_replication_test.go:975: Verified IsCloseTransferTaskAcked
and IsForceReplication flags in SyncWorkflowStateTask`
diff --git a/api/historyservice/v1/request_response.pb.go b/api/historyservice/v1/request_response.pb.go
diff --git a/api/replication/v1/message.pb.go b/api/replication/v1/message.pb.go
diff --git a/common/persistence/serialization/task_serializer.go b/common/persistence/serialization/task_serializer.go
@@ -1293,15 +1293,16 @@ func (s *TaskSerializer) replicationSyncWorkflowStateTaskToProto(
 	syncWorkflowStateTask *tasks.SyncWorkflowStateTask,
 ) *persistencespb.ReplicationTaskInfo {
 	return &persistencespb.ReplicationTaskInfo{
-		NamespaceId:    syncWorkflowStateTask.WorkflowKey.NamespaceID,
-		WorkflowId:     syncWorkflowStateTask.WorkflowKey.WorkflowID,
-		RunId:          syncWorkflowStateTask.WorkflowKey.RunID,
-		TaskType:       enumsspb.TASK_TYPE_REPLICATION_SYNC_WORKFLOW_STATE,
-		TaskId:         syncWorkflowStateTask.TaskID,
-		Version:        syncWorkflowStateTask.Version,
-		VisibilityTime: timestamppb.New(syncWorkflowStateTask.VisibilityTimestamp),
-		Priority:       syncWorkflowStateTask.Priority,
-		TargetClusters: syncWorkflowStateTask.TargetClusters,
+		NamespaceId:        syncWorkflowStateTask.NamespaceID,
+		WorkflowId:         syncWorkflowStateTask.WorkflowID,
+		RunId:              syncWorkflowStateTask.RunID,
+		TaskType:           enumsspb.TASK_TYPE_REPLICATION_SYNC_WORKFLOW_STATE,
+		TaskId:             syncWorkflowStateTask.TaskID,
+		Version:            syncWorkflowStateTask.Version,
+		VisibilityTime:     timestamppb.New(syncWorkflowStateTask.VisibilityTimestamp),
+		Priority:           syncWorkflowStateTask.Priority,
+		TargetClusters:     syncWorkflowStateTask.TargetClusters,
+		IsForceReplication: syncWorkflowStateTask.IsForceReplication,
 	}
 }
 
@@ -1323,6 +1324,7 @@ func (s *TaskSerializer) replicationSyncWorkflowStateTaskFromProto(
 		TaskID:              syncWorkflowStateTask.TaskId,
 		Priority:            syncWorkflowStateTask.Priority,
 		TargetClusters:      syncWorkflowStateTask.TargetClusters,
+		IsForceReplication:  syncWorkflowStateTask.IsForceReplication,
 	}
 }
 
diff --git a/proto/internal/temporal/server/api/historyservice/v1/request_response.proto b/proto/internal/temporal/server/api/historyservice/v1/request_response.proto
@@ -606,6 +606,8 @@ message ReplicateWorkflowStateRequest {
     temporal.server.api.persistence.v1.WorkflowMutableState workflow_state = 1;
     string remote_cluster = 2;
     string namespace_id= 3;
+    bool is_force_replication = 4;
+    bool is_close_transfer_task_acked = 5;
 }
 
 message ReplicateWorkflowStateResponse {
diff --git a/proto/internal/temporal/server/api/replication/v1/message.proto b/proto/internal/temporal/server/api/replication/v1/message.proto
@@ -179,6 +179,8 @@ message HistoryTaskAttributes {
 
 message SyncWorkflowStateTaskAttributes {
     temporal.server.api.persistence.v1.WorkflowMutableState workflow_state = 1;
+    bool is_force_replication = 2;
+    bool is_close_transfer_task_acked = 3;
 }
 
 message TaskQueueUserDataAttributes {
diff --git a/service/history/ndc/workflow_state_replicator.go b/service/history/ndc/workflow_state_replicator.go
@@ -187,7 +187,8 @@ func (r *WorkflowStateReplicatorImpl) SyncWorkflowState(
 	default:
 		return err
 	}
-	return r.applySnapshotWhenWorkflowNotExist(ctx, namespaceID, wid, rid, wfCtx, releaseFn, request.GetWorkflowState(), request.RemoteCluster, nil, false, false)
+	skipCloseTransferTask := request.GetIsForceReplication() && request.GetIsCloseTransferTaskAcked()
+	return r.applySnapshotWhenWorkflowNotExist(ctx, namespaceID, wid, rid, wfCtx, releaseFn, request.GetWorkflowState(), request.RemoteCluster, nil, false, skipCloseTransferTask)
 }
 
 //nolint:revive // cognitive complexity 37 (> max enabled 25)
diff --git a/service/history/replication/executable_workflow_state_task.go b/service/history/replication/executable_workflow_state_task.go
@@ -62,9 +62,11 @@ func NewExecutableWorkflowStateTask(
 			replicationTask,
 		),
 		req: &historyservice.ReplicateWorkflowStateRequest{
-			NamespaceId:   namespaceID,
-			WorkflowState: task.GetWorkflowState(),
-			RemoteCluster: sourceClusterName,
+			NamespaceId:              namespaceID,
+			WorkflowState:            task.GetWorkflowState(),
+			RemoteCluster:            sourceClusterName,
+			IsForceReplication:       task.GetIsForceReplication(),
+			IsCloseTransferTaskAcked: task.GetIsCloseTransferTaskAcked(),
 		},
 	}
 }
diff --git a/service/history/replication/raw_task_converter.go b/service/history/replication/raw_task_converter.go
@@ -223,13 +223,21 @@ func convertWorkflowStateReplicationTask(
 			if err := common.DiscardUnknownProto(workflowMutableState); err != nil {
 				return nil, err
 			}
+
+			isCloseTransferTaskAcked := isCloseTransferTaskAckedForWorkflow(shardContext, &tasks.CloseExecutionTask{
+				WorkflowKey: mutableState.GetWorkflowKey(),
+				TaskID:      mutableState.GetExecutionInfo().GetCloseTransferTaskId(),
+			})
+
 			return &replicationspb.ReplicationTask{
 				TaskType:     enumsspb.REPLICATION_TASK_TYPE_SYNC_WORKFLOW_STATE_TASK,
 				SourceTaskId: taskInfo.TaskID,
 				Priority:     taskInfo.Priority,
 				Attributes: &replicationspb.ReplicationTask_SyncWorkflowStateTaskAttributes{
 					SyncWorkflowStateTaskAttributes: &replicationspb.SyncWorkflowStateTaskAttributes{
-						WorkflowState: workflowMutableState,
+						WorkflowState:            workflowMutableState,
+						IsForceReplication:       taskInfo.IsForceReplication,
+						IsCloseTransferTaskAcked: isCloseTransferTaskAcked,
 					},
 				},
 				VisibilityTime: timestamppb.New(taskInfo.VisibilityTimestamp),
@@ -864,12 +872,19 @@ func (c *syncVersionedTransitionTaskConverter) generateBackfillHistoryTask(
 
 func (c *syncVersionedTransitionTaskConverter) isCloseTransferTaskAcked(
 	closeTransferTask *tasks.CloseExecutionTask,
+) bool {
+	return isCloseTransferTaskAckedForWorkflow(c.shardContext, closeTransferTask)
+}
+
+func isCloseTransferTaskAckedForWorkflow(
+	shardContext historyi.ShardContext,
+	closeTransferTask *tasks.CloseExecutionTask,
 ) bool {
 	if closeTransferTask.TaskID == 0 {
 		return false
 	}
 
-	transferQueueState, ok := c.shardContext.GetQueueState(tasks.CategoryTransfer)
+	transferQueueState, ok := shardContext.GetQueueState(tasks.CategoryTransfer)
 	if !ok {
 		return false
 	}
diff --git a/service/history/replication/raw_task_converter_test.go b/service/history/replication/raw_task_converter_test.go
@@ -560,6 +560,16 @@ func (s *rawTaskConverterSuite) TestConvertWorkflowStateReplicationTask_Workflow
 		},
 	}).AnyTimes()
 	s.mutableState.EXPECT().GetWorkflowStateStatus().Return(enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED).AnyTimes()
+	// Mock for watermark check
+	executionInfo := &persistencespb.WorkflowExecutionInfo{
+		NamespaceId:                       s.namespaceID,
+		WorkflowId:                        s.workflowID,
+		TaskGenerationShardClockTimestamp: 123,
+		CloseVisibilityTaskId:             456,
+		CloseTransferTaskId:               789,
+	}
+	s.mutableState.EXPECT().GetExecutionInfo().Return(executionInfo).AnyTimes()
+	s.mutableState.EXPECT().GetWorkflowKey().Return(definition.NewWorkflowKey(s.namespaceID, s.workflowID, s.runID)).AnyTimes()
 
 	result, err := convertWorkflowStateReplicationTask(ctx, s.shardContext, task, s.workflowCache)
 	s.NoError(err)
@@ -571,7 +581,9 @@ func (s *rawTaskConverterSuite) TestConvertWorkflowStateReplicationTask_Workflow
 		SourceTaskId: task.TaskID,
 		Attributes: &replicationspb.ReplicationTask_SyncWorkflowStateTaskAttributes{
 			SyncWorkflowStateTaskAttributes: &replicationspb.SyncWorkflowStateTaskAttributes{
-				WorkflowState: sanitizedMutableState,
+				WorkflowState:            sanitizedMutableState,
+				IsForceReplication:       task.IsForceReplication,
+				IsCloseTransferTaskAcked: false, // No queue state available
 			},
 		},
 		VisibilityTime: timestamppb.New(task.VisibilityTimestamp),
diff --git a/service/history/tasks/sync_workflow_state_task.go b/service/history/tasks/sync_workflow_state_task.go
@@ -15,9 +15,10 @@ type (
 		VisibilityTimestamp time.Time
 		TaskID              int64
 		// TODO: validate this version in source task converter
-		Version        int64
-		Priority       enumsspb.TaskPriority
-		TargetClusters []string
+		Version            int64
+		Priority           enumsspb.TaskPriority
+		TargetClusters     []string
+		IsForceReplication bool
 	}
 )
 
diff --git a/service/history/workflow/task_generator.go b/service/history/workflow/task_generator.go
@@ -755,10 +755,11 @@ func (r *TaskGeneratorImpl) GenerateMigrationTasks(targetClusters []string) ([]t
 	if r.mutableState.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
 		syncWorkflowStateTask := []tasks.Task{&tasks.SyncWorkflowStateTask{
 			// TaskID, VisibilityTimestamp is set by shard
-			WorkflowKey:    workflowKey,
-			Version:        lastItem.GetVersion(),
-			Priority:       enumsspb.TASK_PRIORITY_LOW,
-			TargetClusters: targetClusters,
+			WorkflowKey:        workflowKey,
+			Version:            lastItem.GetVersion(),
+			Priority:           enumsspb.TASK_PRIORITY_LOW,
+			TargetClusters:     targetClusters,
+			IsForceReplication: true,
 		}}
 		if r.mutableState.IsTransitionHistoryEnabled() &&
 			// even though current cluster may enabled state transition, but transition history can be cleared
diff --git a/tests/xdc/stream_based_replication_test.go b/tests/xdc/stream_based_replication_test.go
@@ -779,9 +779,8 @@ func (s *streamBasedReplicationTestSuite) TestResetWorkflow_SyncWorkflowState()
 }
 
 func (s *streamBasedReplicationTestSuite) TestCloseTransferTaskAckedReplication() {
-	if !s.enableTransitionHistory {
-		s.T().Skip("Skip when transition history is disabled")
-	}
+	// Test works for both SyncVersionedTransitionTask (with transition history)
+	// and SyncWorkflowStateTask (without transition history)
 	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
 	defer cancel()
 
@@ -920,31 +919,61 @@ func (s *streamBasedReplicationTestSuite) TestCloseTransferTaskAckedReplication(
 	s.T().Log("Generated last history replication tasks to force replication of completed workflow")
 
 	recorder := s.clusters[0].GetReplicationStreamRecorder()
-	s.T().Log("Checking replication stream for close transfer task acknowledgment in versioned transition artifact...")
-	s.Eventually(func() bool {
-		for _, msg := range recorder.GetMessages() {
-			if msg.Direction != testcore.DirectionServerSend {
-				continue
-			}
 
-			resp := testcore.ExtractReplicationMessages(msg.Request)
-			if resp == nil {
-				continue
+	// Check for flags based on transition history mode
+	if s.enableTransitionHistory {
+		// With transition history: check SyncVersionedTransitionTask
+		s.T().Log("Checking replication stream for close transfer task acknowledgment in versioned transition artifact...")
+		s.Eventually(func() bool {
+			for _, msg := range recorder.GetMessages() {
+				if msg.Direction != testcore.DirectionServerSend {
+					continue
+				}
+
+				resp := testcore.ExtractReplicationMessages(msg.Request)
+				if resp == nil {
+					continue
+				}
+
+				for _, task := range resp.GetReplicationTasks() {
+					if syncAttrs := task.GetSyncVersionedTransitionTaskAttributes(); syncAttrs != nil {
+						if artifact := syncAttrs.GetVersionedTransitionArtifact(); artifact != nil {
+							if artifact.GetIsCloseTransferTaskAcked() && artifact.GetIsForceReplication() {
+								return true
+							}
+						}
+					}
+				}
 			}
+			return false
+		}, 10*time.Second, 100*time.Millisecond)
+		s.T().Log("Verified IsCloseTransferTaskAcked and IsForceReplication flags in SyncVersionedTransitionTask")
+	} else {
+		// Without transition history: check SyncWorkflowStateTask
+		s.T().Log("Checking replication stream for close transfer task acknowledgment in workflow state attributes...")
+		s.Eventually(func() bool {
+			for _, msg := range recorder.GetMessages() {
+				if msg.Direction != testcore.DirectionServerSend {
+					continue
+				}
+
+				resp := testcore.ExtractReplicationMessages(msg.Request)
+				if resp == nil {
+					continue
+				}
 
-			for _, task := range resp.GetReplicationTasks() {
-				if syncAttrs := task.GetSyncVersionedTransitionTaskAttributes(); syncAttrs != nil {
-					if artifact := syncAttrs.GetVersionedTransitionArtifact(); artifact != nil {
-						if artifact.GetIsCloseTransferTaskAcked() && artifact.GetIsForceReplication() {
+				for _, task := range resp.GetReplicationTasks() {
+					if workflowStateAttrs := task.GetSyncWorkflowStateTaskAttributes(); workflowStateAttrs != nil {
+						if workflowStateAttrs.GetIsCloseTransferTaskAcked() && workflowStateAttrs.GetIsForceReplication() {
 							return true
 						}
 					}
 				}
 			}
-		}
-		return false
-	}, 10*time.Second, 100*time.Millisecond)
-	s.T().Log("Verified IsCloseTransferTaskAcked flag is set in replication artifact")
+			return false
+		}, 10*time.Second, 100*time.Millisecond)
+		s.T().Log("Verified IsCloseTransferTaskAcked and IsForceReplication flags in SyncWorkflowStateTask")
+	}
 
 	// Wait for replication to complete to the passive cluster
 	s.T().Log("Waiting for workflow to replicate to cluster 1 (passive)...")

Original file line number	Diff line number	Diff line change
`@@ -606,6 +606,8 @@ message ReplicateWorkflowStateRequest {`
`606`	`606`	`temporal.server.api.persistence.v1.WorkflowMutableState workflow_state = 1;`
`607`	`607`	`string remote_cluster = 2;`
`608`	`608`	`string namespace_id= 3;`
	`609`	`+ bool is_force_replication = 4;`
	`610`	`+ bool is_close_transfer_task_acked = 5;`
`609`	`611`	`}`
`610`	`612`
`611`	`613`	`message ReplicateWorkflowStateResponse {`
Original file line number	Diff line number	Diff line change
`@@ -179,6 +179,8 @@ message HistoryTaskAttributes {`
`179`	`179`
`180`	`180`	`message SyncWorkflowStateTaskAttributes {`
`181`	`181`	`temporal.server.api.persistence.v1.WorkflowMutableState workflow_state = 1;`
	`182`	`+ bool is_force_replication = 2;`
	`183`	`+ bool is_close_transfer_task_acked = 3;`
`182`	`184`	`}`
`183`	`185`
`184`	`186`	`message TaskQueueUserDataAttributes {`
Original file line number	Diff line number	Diff line change
`@@ -187,7 +187,8 @@ func (r *WorkflowStateReplicatorImpl) SyncWorkflowState(`
`187`	`187`	`default:`
`188`	`188`	`return err`
`189`	`189`	`}`
`190`		`- return r.applySnapshotWhenWorkflowNotExist(ctx, namespaceID, wid, rid, wfCtx, releaseFn, request.GetWorkflowState(), request.RemoteCluster, nil, false, false)`
	`190`	`+ skipCloseTransferTask := request.GetIsForceReplication() && request.GetIsCloseTransferTaskAcked()`
	`191`	`+ return r.applySnapshotWhenWorkflowNotExist(ctx, namespaceID, wid, rid, wfCtx, releaseFn, request.GetWorkflowState(), request.RemoteCluster, nil, false, skipCloseTransferTask)`
`191`	`192`	`}`
`192`	`193`
`193`	`194`	`//nolint:revive // cognitive complexity 37 (> max enabled 25)`
Original file line number	Diff line number	Diff line change
`@@ -15,9 +15,10 @@ type (`
`15`	`15`	`VisibilityTimestamp time.Time`
`16`	`16`	`TaskID int64`
`17`	`17`	`// TODO: validate this version in source task converter`
`18`		`- Version int64`
`19`		`- Priority enumsspb.TaskPriority`
`20`		`- TargetClusters []string`
	`18`	`+ Version int64`
	`19`	`+ Priority enumsspb.TaskPriority`
	`20`	`+ TargetClusters []string`
	`21`	`+ IsForceReplication bool`
`21`	`22`	`}`
`22`	`23`	`)`
`23`	`24`