Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
## 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`
This commit is contained in:
michaely520
2025-10-28 11:55:52 -07:00
committed by GitHub
parent 661ad8ce7c
commit 70dcf17262
12 changed files with 161 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
api/historyservice/v1/request_response.pb.go
View File

@@ -4141,12 +4141,14 @@ func (*ReplicateEventsV2Response) Descriptor() ([]byte, []int) {
}
type ReplicateWorkflowStateRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
WorkflowState *v19.WorkflowMutableState `protobuf:"bytes,1,opt,name=workflow_state,json=workflowState,proto3" json:"workflow_state,omitempty"`
RemoteCluster string `protobuf:"bytes,2,opt,name=remote_cluster,json=remoteCluster,proto3" json:"remote_cluster,omitempty"`
NamespaceId string `protobuf:"bytes,3,opt,name=namespace_id,json=namespaceId,proto3" json:"namespace_id,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
state protoimpl.MessageState `protogen:"open.v1"`
WorkflowState *v19.WorkflowMutableState `protobuf:"bytes,1,opt,name=workflow_state,json=workflowState,proto3" json:"workflow_state,omitempty"`
RemoteCluster string `protobuf:"bytes,2,opt,name=remote_cluster,json=remoteCluster,proto3" json:"remote_cluster,omitempty"`
NamespaceId string `protobuf:"bytes,3,opt,name=namespace_id,json=namespaceId,proto3" json:"namespace_id,omitempty"`
IsForceReplication bool `protobuf:"varint,4,opt,name=is_force_replication,json=isForceReplication,proto3" json:"is_force_replication,omitempty"`
IsCloseTransferTaskAcked bool `protobuf:"varint,5,opt,name=is_close_transfer_task_acked,json=isCloseTransferTaskAcked,proto3" json:"is_close_transfer_task_acked,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ReplicateWorkflowStateRequest) Reset() {
@@ -4200,6 +4202,20 @@ func (x *ReplicateWorkflowStateRequest) GetNamespaceId() string {
return ""
}
func (x *ReplicateWorkflowStateRequest) GetIsForceReplication() bool {
if x != nil {
return x.IsForceReplication
}
return false
}
func (x *ReplicateWorkflowStateRequest) GetIsCloseTransferTaskAcked() bool {
if x != nil {
return x.IsCloseTransferTaskAcked
}
return false
}
type ReplicateWorkflowStateResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
unknownFields protoimpl.UnknownFields
@@ -10122,11 +10138,13 @@ const file_temporal_server_api_historyservice_v1_request_response_proto_rawDesc
"\x13base_execution_info\x18\x06 \x01(\v22.temporal.server.api.workflow.v1.BaseExecutionInfoR\x11baseExecutionInfo\x12\x1c\n" +
"\n" +
"new_run_id\x18\a \x01(\tR\bnewRunId:$\x92\xc4\x03 *\x1eworkflow_execution.workflow_id\"\x1b\n" +
"\x19ReplicateEventsV2Response\"\xfb\x01\n" +
"\x19ReplicateEventsV2Response\"\xed\x02\n" +
"\x1dReplicateWorkflowStateRequest\x12_\n" +
"\x0eworkflow_state\x18\x01 \x01(\v28.temporal.server.api.persistence.v1.WorkflowMutableStateR\rworkflowState\x12%\n" +
"\x0eremote_cluster\x18\x02 \x01(\tR\rremoteCluster\x12!\n" +
"\fnamespace_id\x18\x03 \x01(\tR\vnamespaceId:/\x92\xc4\x03+*)workflow_state.execution_info.workflow_id\" \n" +
"\fnamespace_id\x18\x03 \x01(\tR\vnamespaceId\x120\n" +
"\x14is_force_replication\x18\x04 \x01(\bR\x12isForceReplication\x12>\n" +
"\x1cis_close_transfer_task_acked\x18\x05 \x01(\bR\x18isCloseTransferTaskAcked:/\x92\xc4\x03+*)workflow_state.execution_info.workflow_id\" \n" +
"\x1eReplicateWorkflowStateResponse\"\xa7\x01\n" +
"\x16SyncShardStatusRequest\x12%\n" +
"\x0esource_cluster\x18\x01 \x01(\tR\rsourceCluster\x12\x19\n" +

30
api/replication/v1/message.pb.go
View File

@@ -1325,10 +1325,12 @@ func (x *HistoryTaskAttributes) GetEventsBatches() []*v11.DataBlob {
}
type SyncWorkflowStateTaskAttributes struct {
state protoimpl.MessageState `protogen:"open.v1"`
WorkflowState *v12.WorkflowMutableState `protobuf:"bytes,1,opt,name=workflow_state,json=workflowState,proto3" json:"workflow_state,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
state protoimpl.MessageState `protogen:"open.v1"`
WorkflowState *v12.WorkflowMutableState `protobuf:"bytes,1,opt,name=workflow_state,json=workflowState,proto3" json:"workflow_state,omitempty"`
IsForceReplication bool `protobuf:"varint,2,opt,name=is_force_replication,json=isForceReplication,proto3" json:"is_force_replication,omitempty"`
IsCloseTransferTaskAcked bool `protobuf:"varint,3,opt,name=is_close_transfer_task_acked,json=isCloseTransferTaskAcked,proto3" json:"is_close_transfer_task_acked,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *SyncWorkflowStateTaskAttributes) Reset() {
@@ -1368,6 +1370,20 @@ func (x *SyncWorkflowStateTaskAttributes) GetWorkflowState() *v12.WorkflowMutabl
return nil
}
func (x *SyncWorkflowStateTaskAttributes) GetIsForceReplication() bool {
if x != nil {
return x.IsForceReplication
}
return false
}
func (x *SyncWorkflowStateTaskAttributes) GetIsCloseTransferTaskAcked() bool {
if x != nil {
return x.IsCloseTransferTaskAcked
}
return false
}
type TaskQueueUserDataAttributes struct {
state protoimpl.MessageState `protogen:"open.v1"`
NamespaceId string `protobuf:"bytes,1,opt,name=namespace_id,json=namespaceId,proto3" json:"namespace_id,omitempty"`
@@ -2133,9 +2149,11 @@ const file_temporal_server_api_replication_v1_message_proto_rawDesc = "" +
"\n" +
"new_run_id\x18\t \x01(\tR\bnewRunId\x12G\n" +
"\x0eevents_batches\x18\n" +
" \x03(\v2 .temporal.api.common.v1.DataBlobR\reventsBatchesJ\x04\b\x01\x10\x02\"\x82\x01\n" +
" \x03(\v2 .temporal.api.common.v1.DataBlobR\reventsBatchesJ\x04\b\x01\x10\x02\"\xf4\x01\n" +
"\x1fSyncWorkflowStateTaskAttributes\x12_\n" +
"\x0eworkflow_state\x18\x01 \x01(\v28.temporal.server.api.persistence.v1.WorkflowMutableStateR\rworkflowState\"\xbc\x01\n" +
"\x0eworkflow_state\x18\x01 \x01(\v28.temporal.server.api.persistence.v1.WorkflowMutableStateR\rworkflowState\x120\n" +
"\x14is_force_replication\x18\x02 \x01(\bR\x12isForceReplication\x12>\n" +
"\x1cis_close_transfer_task_acked\x18\x03 \x01(\bR\x18isCloseTransferTaskAcked\"\xbc\x01\n" +
"\x1bTaskQueueUserDataAttributes\x12!\n" +
"\fnamespace_id\x18\x01 \x01(\tR\vnamespaceId\x12&\n" +
"\x0ftask_queue_name\x18\x02 \x01(\tR\rtaskQueueName\x12R\n" +

20
common/persistence/serialization/task_serializer.go
View File

@@ -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,
}
}

2
proto/internal/temporal/server/api/historyservice/v1/request_response.proto
View File

@@ -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 {

2
proto/internal/temporal/server/api/replication/v1/message.proto
View File

@@ -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 {

3
service/history/ndc/workflow_state_replicator.go
View File

@@ -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)

8
service/history/replication/executable_workflow_state_task.go
View File

@@ -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(),
},
}
}

19
service/history/replication/raw_task_converter.go
View File

@@ -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
}

14
service/history/replication/raw_task_converter_test.go
View File

@@ -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),

7
service/history/tasks/sync_workflow_state_task.go
View File

@@ -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
}
)

9
service/history/workflow/task_generator.go
View File

@@ -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

71
tests/xdc/stream_based_replication_test.go
View File

@@ -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
}
for _, task := range resp.GetReplicationTasks() {
if syncAttrs := task.GetSyncVersionedTransitionTaskAttributes(); syncAttrs != nil {
if artifact := syncAttrs.GetVersionedTransitionArtifact(); artifact != nil {
if artifact.GetIsCloseTransferTaskAcked() && artifact.GetIsForceReplication() {
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 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)...")