Replicate workflow deleteion (#9717)

## What changed? Add a new replication task type `DeleteExecutionReplicationTask` that replicates workflow deletion from the active cluster to passive/standby clusters. Gated by feature flag `history.enableDeleteWorkflowExecutionReplication` (default: false). #### **Key changes across the replication pipeline:** 1. Proto enums: `TASK_TYPE_REPLICATION_DELETE_EXECUTION` (34), `REPLICATION_TASK_TYPE_DELETE_EXECUTION_TASK` (13) 2. Replication task is associated with a new stage in `ShardContext.DeleteWorkflowExecution`, bundled with delete visibility task. 3. ~Engine interface: added `ForceDeleteWorkflowExecution` so the task can invoke the `ForceDeleteWorkflowExecution`.~ ## Why? Today, when a user delete workflow execution in source cluster, this operation will not replicate to the standby/target clusters. When a namespace failover to a target cluster, those deleted workflow may resurrected. <details> <summary>Race condition analysis</summary> **Before this change:** 1. **Cross-cluster resurrection:** Active deletes workflow → standby untouched → failover → workflow reappears. 2. **Termination event silently dropped:** Deleting a running workflow terminates it first, generating a `HistoryReplicationTask`. But the async `CloseExecutionTask` may delete mutable state before the stream sender converts that task. The converter calls `getBranchToken()` → `NotFound` → task silently dropped. The standby never sees the termination or the deletion. **After this change:** Race 1 is fixed — `DeleteExecutionReplicationTask` explicitly tells the standby to delete. Race 2 is mitigated — even if the termination event's replication task is dropped, the delete replication task ensures the standby cleans up. - If the workflow is still running (termination not yet replicated), the `DeleteExecutionTask` reschedules itself until the workflow closes. - If the termination event arrives later, the workflow closes normally, then the delete proceeds. - If the workflow is already deleted (e.g., by retention), the task is a no-op (`NotFound` treated as success). </details> <details> <summary>Deletion paths</summary> | Path | Replication task? | |------|-------------------| | User deletes workflow (active, running or closed) | Yes | | User deletes on passive (DC forwarding ON) | Forwarded to active → yes | | User deletes on passive (no forwarding) | No — `ActiveInCluster` check skips | | Retention expiry (with or without archival) | No — stage pre-marked as processed | | Admin ForceDelete (tdbg) | No — bypasses `DeleteWorkflowExecution` | </details> ## How did you test it? - [x] built - [x] run locally and tested manually - [ ] covered by existing tests - [x] added new unit test(s) - [x] added new functional test(s) Before change: <img width="1507" height="163" alt="Screenshot 2026-03-26 at 11 59 51 PM" src="https://github.com/user-attachments/assets/118cc50e-b69d-468a-9e45-5f49e4e4b9d1" /> After change: <img width="1507" height="135" alt="Screenshot 2026-03-27 at 12 00 07 AM" src="https://github.com/user-attachments/assets/8ccb7a11-2cb4-48b5-af89-b4a60ddb6333" /> ## Potential risks n/a
2026-08-30 18:41:49 -07:00 · 2026-04-02 14:49:09 -07:00
parent 3c84830187
commit 72567b9ed1
24 changed files with 873 additions and 38 deletions
--- a/api/enums/v1/replication.go-helpers.pb.go
+++ b/api/enums/v1/replication.go-helpers.pb.go
@@ -20,6 +20,7 @@ var (
 		"BackfillHistoryTask":           10,
 		"VerifyVersionedTransitionTask": 11,
 		"SyncVersionedTransitionTask":   12,
+		"DeleteExecutionTask":           13,
 	}
 )

--- a/api/enums/v1/replication.pb.go
+++ b/api/enums/v1/replication.pb.go
@@ -39,6 +39,7 @@ const (
 	REPLICATION_TASK_TYPE_BACKFILL_HISTORY_TASK            ReplicationTaskType = 10
 	REPLICATION_TASK_TYPE_VERIFY_VERSIONED_TRANSITION_TASK ReplicationTaskType = 11
 	REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK   ReplicationTaskType = 12
+	REPLICATION_TASK_TYPE_DELETE_EXECUTION_TASK            ReplicationTaskType = 13
 )

 // Enum value maps for ReplicationTaskType.
@@ -57,6 +58,7 @@ var (
 		10: "REPLICATION_TASK_TYPE_BACKFILL_HISTORY_TASK",
 		11: "REPLICATION_TASK_TYPE_VERIFY_VERSIONED_TRANSITION_TASK",
 		12: "REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK",
+		13: "REPLICATION_TASK_TYPE_DELETE_EXECUTION_TASK",
 	}
 	ReplicationTaskType_value = map[string]int32{
 		"REPLICATION_TASK_TYPE_UNSPECIFIED":                      0,
@@ -72,6 +74,7 @@ var (
 		"REPLICATION_TASK_TYPE_BACKFILL_HISTORY_TASK":            10,
 		"REPLICATION_TASK_TYPE_VERIFY_VERSIONED_TRANSITION_TASK": 11,
 		"REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK":   12,
+		"REPLICATION_TASK_TYPE_DELETE_EXECUTION_TASK":            13,
 	}
 )

@@ -111,11 +114,12 @@ func (x ReplicationTaskType) String() string {
 		return "VerifyVersionedTransitionTask"
 	case REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK:
 		return "SyncVersionedTransitionTask"
-	default:
-		return strconv.
+	case REPLICATION_TASK_TYPE_DELETE_EXECUTION_TASK:

-			// Enum value maps for NamespaceOperation.
-			Itoa(int(x))
+		// Enum value maps for NamespaceOperation.
+		return "DeleteExecutionTask"
+	default:
+		return strconv.Itoa(int(x))
 	}

 }
@@ -257,7 +261,7 @@ var File_temporal_server_api_enums_v1_replication_proto protoreflect.FileDescrip

 const file_temporal_server_api_enums_v1_replication_proto_rawDesc = "" +
 	"\n" +
-	".temporal/server/api/enums/v1/replication.proto\x12\x1ctemporal.server.api.enums.v1*\xfe\x04\n" +
+	".temporal/server/api/enums/v1/replication.proto\x12\x1ctemporal.server.api.enums.v1*\xaf\x05\n" +
 	"\x13ReplicationTaskType\x12%\n" +
 	"!REPLICATION_TASK_TYPE_UNSPECIFIED\x10\x00\x12(\n" +
 	"$REPLICATION_TASK_TYPE_NAMESPACE_TASK\x10\x01\x12&\n" +
@@ -272,7 +276,8 @@ const file_temporal_server_api_enums_v1_replication_proto_rawDesc = "" +
 	"+REPLICATION_TASK_TYPE_BACKFILL_HISTORY_TASK\x10\n" +
 	"\x12:\n" +
 	"6REPLICATION_TASK_TYPE_VERIFY_VERSIONED_TRANSITION_TASK\x10\v\x128\n" +
-	"4REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK\x10\f*y\n" +
+	"4REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK\x10\f\x12/\n" +
+	"+REPLICATION_TASK_TYPE_DELETE_EXECUTION_TASK\x10\r*y\n" +
 	"\x12NamespaceOperation\x12#\n" +
 	"\x1fNAMESPACE_OPERATION_UNSPECIFIED\x10\x00\x12\x1e\n" +
 	"\x1aNAMESPACE_OPERATION_CREATE\x10\x01\x12\x1e\n" +
--- a/api/enums/v1/task.go-helpers.pb.go
+++ b/api/enums/v1/task.go-helpers.pb.go
@@ -57,6 +57,7 @@ var (
 		"ReplicationSyncVersionedTransition": 31,
 		"ChasmPure":                          32,
 		"Chasm":                              33,
+		"ReplicationDeleteExecution":         34,
 	}
 )

--- a/api/enums/v1/task.pb.go
+++ b/api/enums/v1/task.pb.go
@@ -126,6 +126,8 @@ const (
 	TASK_TYPE_CHASM_PURE TaskType = 32
 	// A task with side effects generated by a CHASM component.
 	TASK_TYPE_CHASM TaskType = 33
+	// A replication task that deletes workflow on passive cluster(s).
+	TASK_TYPE_REPLICATION_DELETE_EXECUTION TaskType = 34
 )

 // Enum value maps for TaskType.
@@ -162,6 +164,7 @@ var (
 		31: "TASK_TYPE_REPLICATION_SYNC_VERSIONED_TRANSITION",
 		32: "TASK_TYPE_CHASM_PURE",
 		33: "TASK_TYPE_CHASM",
+		34: "TASK_TYPE_REPLICATION_DELETE_EXECUTION",
 	}
 	TaskType_value = map[string]int32{
 		"TASK_TYPE_UNSPECIFIED":                           0,
@@ -195,6 +198,7 @@ var (
 		"TASK_TYPE_REPLICATION_SYNC_VERSIONED_TRANSITION": 31,
 		"TASK_TYPE_CHASM_PURE":                            32,
 		"TASK_TYPE_CHASM":                                 33,
+		"TASK_TYPE_REPLICATION_DELETE_EXECUTION":          34,
 	}
 )

@@ -278,6 +282,8 @@ func (x TaskType) String() string {
 		return "ChasmPure"
 	case TASK_TYPE_CHASM:
 		return "Chasm"
+	case TASK_TYPE_REPLICATION_DELETE_EXECUTION:
+		return "ReplicationDeleteExecution"
 	default:
 		return strconv.Itoa(int(x))
 	}
@@ -367,7 +373,7 @@ const file_temporal_server_api_enums_v1_task_proto_rawDesc = "" +
 	"TaskSource\x12\x1b\n" +
 	"\x17TASK_SOURCE_UNSPECIFIED\x10\x00\x12\x17\n" +
 	"\x13TASK_SOURCE_HISTORY\x10\x01\x12\x1a\n" +
-	"\x16TASK_SOURCE_DB_BACKLOG\x10\x02*\xb6\t\n" +
+	"\x16TASK_SOURCE_DB_BACKLOG\x10\x02*\xe2\t\n" +
 	"\bTaskType\x12\x19\n" +
 	"\x15TASK_TYPE_UNSPECIFIED\x10\x00\x12!\n" +
 	"\x1dTASK_TYPE_REPLICATION_HISTORY\x10\x01\x12'\n" +
@@ -400,7 +406,8 @@ const file_temporal_server_api_enums_v1_task_proto_rawDesc = "" +
 	"\x1eTASK_TYPE_REPLICATION_SYNC_HSM\x10\x1e\x123\n" +
 	"/TASK_TYPE_REPLICATION_SYNC_VERSIONED_TRANSITION\x10\x1f\x12\x18\n" +
 	"\x14TASK_TYPE_CHASM_PURE\x10 \x12\x13\n" +
-	"\x0fTASK_TYPE_CHASM\x10!\"\x04\b\t\x10\t\"\x04\b\v\x10\v\"\x04\b\x17\x10\x17*\\\n" +
+	"\x0fTASK_TYPE_CHASM\x10!\x12*\n" +
+	"&TASK_TYPE_REPLICATION_DELETE_EXECUTION\x10\"\"\x04\b\t\x10\t\"\x04\b\v\x10\v\"\x04\b\x17\x10\x17*\\\n" +
 	"\fTaskPriority\x12\x1d\n" +
 	"\x19TASK_PRIORITY_UNSPECIFIED\x10\x00\x12\x16\n" +
 	"\x12TASK_PRIORITY_HIGH\x10\x01\x12\x15\n" +
--- a/common/dynamicconfig/constants.go
+++ b/common/dynamicconfig/constants.go
@@ -1528,6 +1528,13 @@ Don't change this on a live cluster without using the gradual change mechanism.
 [go.temporal.io/server/common/persistence.QueueV2]`,
 	)

+	EnableDeleteWorkflowExecutionReplication = NewGlobalBoolSetting(
+		"history.enableDeleteWorkflowExecutionReplication",
+		false,
+		`EnableDeleteWorkflowExecutionReplication controls whether a replication task is generated when a workflow
+execution is deleted. When enabled, workflow deletions on the active cluster will be replicated to passive clusters.`,
+	)
+
 	HistoryRPS = NewGlobalIntSetting(
 		"history.rps",
 		3000,
--- a/common/metrics/metric_defs.go
+++ b/common/metrics/metric_defs.go
@@ -558,6 +558,8 @@ const (
 	UnknownTaskScope = "UnknownTask"
 	// ParentClosePolicyProcessorScope is scope used by all metrics emitted by worker.ParentClosePolicyProcessor
 	ParentClosePolicyProcessorScope = "ParentClosePolicyProcessor"
+	// DeleteExecutionReplicationTaskScope is the scope used by delete execution replication task processing
+	DeleteExecutionReplicationTaskScope = "DeleteExecutionReplicationTask"
 )

 // History task type
@@ -1047,7 +1049,8 @@ var (
 	ReplicationOrphanedHistoryBranch = NewCounterDef("replication_orphaned_history_branch")
 	// ReplicationTasksLag is a heuristic for how far behind the remote DC is for a given cluster. It measures the
 	// difference between task IDs so its unit should be "tasks".
-	ReplicationTasksLag = NewDimensionlessHistogramDef("replication_tasks_lag")
+	ReplicationTasksLag                             = NewDimensionlessHistogramDef("replication_tasks_lag")
+	ReplicationDeleteExecutionTaskGenerationFailure = NewCounterDef("replication_delete_execution_task_generation_failure")
 	// ReplicationTasksFetched records the number of tasks fetched by the poller.
 	ReplicationTasksFetched                        = NewDimensionlessHistogramDef("replication_tasks_fetched")
 	ReplicationLatency                             = NewTimerDef("replication_latency")
--- a/common/persistence/serialization/task_serializers.go
+++ b/common/persistence/serialization/task_serializers.go
@@ -318,6 +318,8 @@ func (t *serializerImpl) DeserializeReplicationTask(replicationTask *persistence
 		return replicationSyncHSMTaskFromProto(replicationTask), nil
 	case enumsspb.TASK_TYPE_REPLICATION_SYNC_VERSIONED_TRANSITION:
 		return replicationSyncVersionedTransitionTaskFromProto(replicationTask, t)
+	case enumsspb.TASK_TYPE_REPLICATION_DELETE_EXECUTION:
+		return replicationDeleteExecutionTaskFromProto(replicationTask), nil
 	default:
 		return nil, serviceerror.NewInternalf("Unknown replication task type: %v", replicationTask.TaskType)
 	}
@@ -335,6 +337,8 @@ func (t *serializerImpl) SerializeReplicationTask(task tasks.Task) (*persistence
 		return replicationSyncHSMTaskToProto(task), nil
 	case *tasks.SyncVersionedTransitionTask:
 		return replicationSyncVersionedTransitionTaskToProto(task, t)
+	case *tasks.DeleteExecutionReplicationTask:
+		return replicationDeleteExecutionTaskToProto(task), nil
 	default:
 		return nil, serviceerror.NewInternalf("Unknown repication task type: %v", task)
 	}
@@ -1492,3 +1496,36 @@ func deserializeOutboundTask(
 		return nil, serviceerror.NewInternalf("unknown outbound task type while deserializing: %v", info)
 	}
 }
+
+func replicationDeleteExecutionTaskToProto(
+	task *tasks.DeleteExecutionReplicationTask,
+) *persistencespb.ReplicationTaskInfo {
+	return &persistencespb.ReplicationTaskInfo{
+		NamespaceId:    task.NamespaceID,
+		WorkflowId:     task.WorkflowID,
+		RunId:          task.RunID,
+		TaskType:       enumsspb.TASK_TYPE_REPLICATION_DELETE_EXECUTION,
+		TaskId:         task.TaskID,
+		VisibilityTime: timestamppb.New(task.VisibilityTimestamp),
+		ArchetypeId:    task.ArchetypeID,
+	}
+}
+
+func replicationDeleteExecutionTaskFromProto(
+	info *persistencespb.ReplicationTaskInfo,
+) *tasks.DeleteExecutionReplicationTask {
+	visibilityTimestamp := time.Unix(0, 0)
+	if info.VisibilityTime != nil {
+		visibilityTimestamp = info.VisibilityTime.AsTime()
+	}
+	return &tasks.DeleteExecutionReplicationTask{
+		WorkflowKey: definition.NewWorkflowKey(
+			info.NamespaceId,
+			info.WorkflowId,
+			info.RunId,
+		),
+		VisibilityTimestamp: visibilityTimestamp,
+		TaskID:              info.TaskId,
+		ArchetypeID:         info.ArchetypeId,
+	}
+}
--- a/common/persistence/serialization/task_serializers_test.go
+++ b/common/persistence/serialization/task_serializers_test.go
@@ -411,6 +411,17 @@ func (s *taskSerializerSuite) TestSyncWorkflowStateTask() {
 	s.assertEqualTasks(syncWorkflowStateTask)
 }

+func (s *taskSerializerSuite) TestDeleteExecutionReplicationTask() {
+	deleteExecutionReplicationTask := &tasks.DeleteExecutionReplicationTask{
+		WorkflowKey:         s.workflowKey,
+		VisibilityTimestamp: time.Unix(0, 0).UTC(), // go == compare for location as well which is striped during marshaling/unmarshaling
+		TaskID:              rand.Int63(),
+		ArchetypeID:         rand.Uint32(),
+	}
+
+	s.assertEqualTasks(deleteExecutionReplicationTask)
+}
+
 func (s *taskSerializerSuite) TestDeleteExecutionTask() {
 	deleteExecutionTask := &tasks.DeleteExecutionTask{
 		WorkflowKey:         s.workflowKey,
--- a/proto/internal/temporal/server/api/enums/v1/replication.proto
+++ b/proto/internal/temporal/server/api/enums/v1/replication.proto
@@ -18,6 +18,7 @@ enum ReplicationTaskType {
  REPLICATION_TASK_TYPE_BACKFILL_HISTORY_TASK = 10;
  REPLICATION_TASK_TYPE_VERIFY_VERSIONED_TRANSITION_TASK = 11;
  REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK = 12;
+  REPLICATION_TASK_TYPE_DELETE_EXECUTION_TASK = 13;
 }

 enum NamespaceOperation {
--- a/proto/internal/temporal/server/api/enums/v1/task.proto
+++ b/proto/internal/temporal/server/api/enums/v1/task.proto
@@ -59,6 +59,9 @@ enum TaskType {

  // A task with side effects generated by a CHASM component.
  TASK_TYPE_CHASM = 33;
+
+  // A replication task that deletes workflow on passive cluster(s).
+  TASK_TYPE_REPLICATION_DELETE_EXECUTION = 34;
 }

 // TaskPriority is only used for replication task as of May 2024
--- a/service/history/configs/config.go
+++ b/service/history/configs/config.go
@@ -316,6 +316,7 @@ type Config struct {
 	ReplicationStreamReceiverLivenessMultiplier         dynamicconfig.IntPropertyFn
 	ReplicationStreamSenderLivenessMultiplier           dynamicconfig.IntPropertyFn
 	EnableHistoryReplicationRateLimiter                 dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	EnableDeleteWorkflowExecutionReplication            dynamicconfig.BoolPropertyFn

 	// The following are used by consistent query
 	MaxBufferedQueryCount dynamicconfig.IntPropertyFn
@@ -616,6 +617,7 @@ func NewConfig(
 		ReplicationStreamReceiverLivenessMultiplier:         dynamicconfig.ReplicationStreamReceiverLivenessMultiplier.Get(dc),
 		ReplicationStreamSenderLivenessMultiplier:           dynamicconfig.ReplicationStreamSenderLivenessMultiplier.Get(dc),
 		EnableHistoryReplicationRateLimiter:                 dynamicconfig.EnableHistoryReplicationRateLimiter.Get(dc),
+		EnableDeleteWorkflowExecutionReplication:            dynamicconfig.EnableDeleteWorkflowExecutionReplication.Get(dc),

 		MaximumBufferedEventsBatch:       dynamicconfig.MaximumBufferedEventsBatch.Get(dc),
 		MaximumBufferedEventsSizeInBytes: dynamicconfig.MaximumBufferedEventsSizeInBytes.Get(dc),
--- a/service/history/deletemanager/delete_manager.go
+++ b/service/history/deletemanager/delete_manager.go
@@ -134,6 +134,9 @@ func (m *DeleteManagerImpl) DeleteWorkflowExecutionByRetention(
 	ms historyi.MutableState,
 	stage *tasks.DeleteWorkflowExecutionStage,
 ) error {
+	// Skip replication for retention-based deletion. Both clusters have independent retention
+	// timers and will delete on their own schedule.
+	stage.MarkProcessed(tasks.DeleteWorkflowExecutionStageReplication)

 	return m.deleteWorkflowExecutionInternal(ctx, nsID, we, weCtx, ms, stage, m.metricsHandler.WithTags(metrics.OperationTag(metrics.HistoryProcessDeleteHistoryEventScope)))
 }
--- a/service/history/deletemanager/delete_manager_test.go
+++ b/service/history/deletemanager/delete_manager_test.go
@@ -164,6 +164,61 @@ func (s *deleteManagerWorkflowSuite) TestDeleteDeletedWorkflowExecution_Error()
 	s.Error(err)
 }

+func (s *deleteManagerWorkflowSuite) TestDeleteWorkflowExecutionByRetention_SkipsReplication() {
+	we := commonpb.WorkflowExecution{
+		WorkflowId: tests.WorkflowID,
+		RunId:      tests.RunID,
+	}
+
+	mockWeCtx := historyi.NewMockWorkflowContext(s.controller)
+	mockMutableState := historyi.NewMockMutableState(s.controller)
+	mockMutableState.EXPECT().GetCurrentBranchToken().Return([]byte{22, 8, 78}, nil)
+	closeExecutionVisibilityTaskID := int64(39)
+	mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
+		CloseVisibilityTaskId: closeExecutionVisibilityTaskID,
+	})
+	mockMutableState.EXPECT().ChasmTree().Return(workflow.NoopChasmTree).AnyTimes()
+	stage := tasks.DeleteWorkflowExecutionStageNone
+
+	s.mockShardContext.EXPECT().DeleteWorkflowExecution(
+		gomock.Any(),
+		definition.WorkflowKey{
+			NamespaceID: tests.NamespaceID.String(),
+			WorkflowID:  tests.WorkflowID,
+			RunID:       tests.RunID,
+		},
+		workflow.NoopChasmTree.ArchetypeID(),
+		[]byte{22, 8, 78},
+		closeExecutionVisibilityTaskID,
+		time.Unix(0, 0).UTC(),
+		gomock.Any(),
+	).DoAndReturn(func(
+		_ context.Context,
+		_ definition.WorkflowKey,
+		_ chasm.ArchetypeID,
+		_ []byte,
+		_ int64,
+		_ time.Time,
+		stagePtr *tasks.DeleteWorkflowExecutionStage,
+	) error {
+		// Verify that replication stage is already marked as processed before DeleteWorkflowExecution is called.
+		s.True(stagePtr.IsProcessed(tasks.DeleteWorkflowExecutionStageReplication),
+			"Replication stage should be pre-marked as processed for retention-based deletion")
+		return nil
+	})
+	mockWeCtx.EXPECT().Clear()
+
+	err := s.deleteManager.DeleteWorkflowExecutionByRetention(
+		context.Background(),
+		tests.NamespaceID,
+		&we,
+		mockWeCtx,
+		mockMutableState,
+		&stage,
+	)
+	s.NoError(err)
+}
+
 func (s *deleteManagerWorkflowSuite) TestDeleteWorkflowExecution_OpenWorkflow() {
 	we := commonpb.WorkflowExecution{
 		WorkflowId: tests.WorkflowID,
--- a/service/history/replication/ack_manager.go
+++ b/service/history/replication/ack_manager.go
@@ -444,6 +444,8 @@ func (p *ackMgrImpl) ConvertTask(
 			task,
 			p.workflowCache,
 		)
+	case *tasks.DeleteExecutionReplicationTask:
+		return convertDeleteExecutionReplicationTask(task)
 	default:
 		return nil, errUnknownReplicationTask
 	}
--- a/service/history/replication/executable_delete_execution_task.go
+++ b/service/history/replication/executable_delete_execution_task.go
@@ -0,0 +1,162 @@
+package replication
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	commonpb "go.temporal.io/api/common/v1"
+	"go.temporal.io/api/serviceerror"
+	enumsspb "go.temporal.io/server/api/enums/v1"
+	"go.temporal.io/server/api/historyservice/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	replicationspb "go.temporal.io/server/api/replication/v1"
+	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/common/definition"
+	"go.temporal.io/server/common/headers"
+	"go.temporal.io/server/common/log/tag"
+	"go.temporal.io/server/common/metrics"
+	"go.temporal.io/server/common/namespace"
+	ctasks "go.temporal.io/server/common/tasks"
+)
+
+type ExecutableDeleteExecutionTask struct {
+	ProcessToolBox
+
+	definition.WorkflowKey
+	ExecutableTask
+}
+
+var _ ctasks.Task = (*ExecutableDeleteExecutionTask)(nil)
+var _ TrackableExecutableTask = (*ExecutableDeleteExecutionTask)(nil)
+
+func NewExecutableDeleteExecutionTask(
+	processToolBox ProcessToolBox,
+	taskID int64,
+	taskCreationTime time.Time,
+	sourceClusterName string,
+	sourceShardKey ClusterShardKey,
+	replicationTask *replicationspb.ReplicationTask,
+) *ExecutableDeleteExecutionTask {
+	task := replicationTask.GetHistoryTaskAttributes()
+
+	return &ExecutableDeleteExecutionTask{
+		ProcessToolBox: processToolBox,
+		WorkflowKey:    definition.NewWorkflowKey(task.NamespaceId, task.WorkflowId, task.RunId),
+		ExecutableTask: NewExecutableTask(
+			processToolBox,
+			taskID,
+			metrics.DeleteExecutionReplicationTaskScope,
+			taskCreationTime,
+			time.Now().UTC(),
+			sourceClusterName,
+			sourceShardKey,
+			replicationTask,
+		),
+	}
+}
+
+func (e *ExecutableDeleteExecutionTask) QueueID() any {
+	return e.WorkflowKey
+}
+
+func (e *ExecutableDeleteExecutionTask) Execute() error {
+	if e.TerminalState() {
+		return nil
+	}
+	e.MarkExecutionStart()
+
+	callerInfo := getReplicaitonCallerInfo(e.GetPriority())
+	namespaceName, apply, err := e.GetNamespaceInfo(headers.SetCallerInfo(
+		context.Background(),
+		callerInfo,
+	), e.NamespaceID, e.WorkflowID)
+	if err != nil {
+		return err
+	} else if !apply {
+		e.Logger.Warn("Skipping the replication task",
+			tag.WorkflowNamespaceID(e.NamespaceID),
+			tag.WorkflowID(e.WorkflowID),
+			tag.WorkflowRunID(e.RunID),
+			tag.TaskID(e.TaskID()),
+		)
+		metrics.ReplicationTasksSkipped.With(e.MetricsHandler).Record(
+			1,
+			metrics.OperationTag(metrics.DeleteExecutionReplicationTaskScope),
+			metrics.NamespaceTag(namespaceName),
+		)
+		return nil
+	}
+
+	// Only process workflow archetype deletion for now.
+	if e.archetypeID() != chasm.WorkflowArchetypeID {
+		return nil
+	}
+
+	ctx, cancel := newTaskContext(namespaceName, e.Config.ReplicationTaskApplyTimeout(), callerInfo)
+	defer cancel()
+
+	shardContext, err := e.ShardController.GetShardByNamespaceWorkflow(
+		namespace.ID(e.NamespaceID),
+		e.WorkflowID,
+	)
+	if err != nil {
+		return err
+	}
+	engine, err := shardContext.GetEngine(ctx)
+	if err != nil {
+		return err
+	}
+
+	_, err = engine.DeleteWorkflowExecution(ctx, &historyservice.DeleteWorkflowExecutionRequest{
+		NamespaceId: e.NamespaceID,
+		WorkflowExecution: &commonpb.WorkflowExecution{
+			WorkflowId: e.WorkflowID,
+			RunId:      e.RunID,
+		},
+	})
+	return err
+}
+
+func (e *ExecutableDeleteExecutionTask) archetypeID() uint32 {
+	if rawInfo := e.ReplicationTask().GetRawTaskInfo(); rawInfo != nil {
+		return rawInfo.ArchetypeId
+	}
+	return chasm.UnspecifiedArchetypeID
+}
+
+func (e *ExecutableDeleteExecutionTask) HandleErr(err error) error {
+	metrics.ReplicationTasksErrorByType.With(e.MetricsHandler).Record(
+		1,
+		metrics.OperationTag(metrics.DeleteExecutionReplicationTaskScope),
+		metrics.NamespaceTag(e.NamespaceName()),
+		metrics.ServiceErrorTypeTag(err),
+	)
+	switch err.(type) {
+	case nil, *serviceerror.NotFound:
+		return nil
+	default:
+		e.Logger.Error("delete execution replication task encountered error",
+			tag.WorkflowNamespaceID(e.NamespaceID),
+			tag.WorkflowID(e.WorkflowID),
+			tag.WorkflowRunID(e.RunID),
+			tag.TaskID(e.TaskID()),
+			tag.Error(err),
+		)
+		return fmt.Errorf("delete execution replication task error: %w", err)
+	}
+}
+
+func (e *ExecutableDeleteExecutionTask) MarkPoisonPill() error {
+	if e.ReplicationTask().GetRawTaskInfo() == nil {
+		e.ReplicationTask().RawTaskInfo = &persistencespb.ReplicationTaskInfo{
+			NamespaceId: e.NamespaceID,
+			WorkflowId:  e.WorkflowID,
+			RunId:       e.RunID,
+			TaskId:      e.TaskID(),
+			TaskType:    enumsspb.TASK_TYPE_REPLICATION_DELETE_EXECUTION,
+		}
+	}
+
+	return e.ExecutableTask.MarkPoisonPill()
+}
--- a/service/history/replication/executable_task_converter.go
+++ b/service/history/replication/executable_task_converter.go
@@ -147,6 +147,15 @@ func (e *executableTaskConverterImpl) convertOne(
 			sourceShardKey,
 			replicationTask,
 		)
+	case enumsspb.REPLICATION_TASK_TYPE_DELETE_EXECUTION_TASK:
+		return NewExecutableDeleteExecutionTask(
+			e.processToolBox,
+			replicationTask.SourceTaskId,
+			taskCreationTime,
+			sourceClusterName,
+			sourceShardKey,
+			replicationTask,
+		)
 	default:
 		e.processToolBox.Logger.Error(fmt.Sprintf("unknown replication task: %v", replicationTask))
 		return NewExecutableUnknownTask(
--- a/service/history/replication/metrics.go
+++ b/service/history/replication/metrics.go
@@ -31,6 +31,8 @@ func TaskOperationTag(
 		return metrics.BackfillHistoryEventsTaskScope
 	case enumsspb.REPLICATION_TASK_TYPE_VERIFY_VERSIONED_TRANSITION_TASK:
 		return metrics.VerifyVersionedTransitionTaskScope
+	case enumsspb.REPLICATION_TASK_TYPE_DELETE_EXECUTION_TASK:
+		return metrics.DeleteExecutionReplicationTaskScope
 	default:
 		return metrics.NoopTaskScope
 	}
@@ -50,6 +52,8 @@ func TaskOperationTagFromTask(
 		return metrics.SyncWorkflowStateTaskScope
 	case enumsspb.TASK_TYPE_REPLICATION_HISTORY:
 		return metrics.HistoryReplicationTaskScope
+	case enumsspb.TASK_TYPE_REPLICATION_DELETE_EXECUTION:
+		return metrics.DeleteExecutionReplicationTaskScope
 	default:
 		return metrics.UnknownTaskScope
 	}
--- a/service/history/replication/raw_task_converter.go
+++ b/service/history/replication/raw_task_converter.go
@@ -321,6 +321,23 @@ func convertSyncVersionedTransitionTask(
 	)
 }

+func convertDeleteExecutionReplicationTask(
+	taskInfo *tasks.DeleteExecutionReplicationTask,
+) (*replicationspb.ReplicationTask, error) {
+	return &replicationspb.ReplicationTask{
+		TaskType:     enumsspb.REPLICATION_TASK_TYPE_DELETE_EXECUTION_TASK,
+		SourceTaskId: taskInfo.TaskID,
+		Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{
+			HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{
+				NamespaceId: taskInfo.NamespaceID,
+				WorkflowId:  taskInfo.WorkflowID,
+				RunId:       taskInfo.RunID,
+			},
+		},
+		VisibilityTime: timestamppb.New(taskInfo.VisibilityTimestamp),
+	}, nil
+}
+
 func convertHistoryReplicationTask(
 	ctx context.Context,
 	shardContext historyi.ShardContext,
--- a/service/history/replication/raw_task_converter_test.go
+++ b/service/history/replication/raw_task_converter_test.go
@@ -1962,3 +1962,29 @@ func (s *rawTaskConverterSuite) TestIsCloseTransferTaskAcked_TaskNotAcked_Contai
 	result := converter.isCloseTransferTaskAcked(closeTransferTask)
 	s.False(result)
 }
+
+func (s *rawTaskConverterSuite) TestConvertDeleteExecutionReplicationTask() {
+	taskID := int64(1444)
+	task := &tasks.DeleteExecutionReplicationTask{
+		WorkflowKey: definition.NewWorkflowKey(
+			s.namespaceID,
+			s.workflowID,
+			s.runID,
+		),
+		VisibilityTimestamp: time.Now().UTC(),
+		TaskID:              taskID,
+		ArchetypeID:         chasm.WorkflowArchetypeID,
+	}
+
+	result, err := convertDeleteExecutionReplicationTask(task)
+	s.NoError(err)
+	s.NotNil(result)
+	s.Equal(enumsspb.REPLICATION_TASK_TYPE_DELETE_EXECUTION_TASK, result.TaskType)
+	s.Equal(taskID, result.SourceTaskId)
+
+	attrs := result.GetHistoryTaskAttributes()
+	s.NotNil(attrs)
+	s.Equal(s.namespaceID, attrs.NamespaceId)
+	s.Equal(s.workflowID, attrs.WorkflowId)
+	s.Equal(s.runID, attrs.RunId)
+}
--- a/service/history/shard/context_impl.go
+++ b/service/history/shard/context_impl.go
@@ -948,7 +948,7 @@ func (s *ContextImpl) DeleteWorkflowExecution(
 	stage *tasks.DeleteWorkflowExecutionStage,
 ) (retErr error) {
 	// DeleteWorkflowExecution is a 4 stages process (order is very important and should not be changed):
-	// 1. Add visibility delete task, i.e. schedule visibility record delete,
+	// 1. Add visibility delete task, i.e. schedule visibility record delete, and execution replication delete task,
 	// 2. Delete current workflow execution pointer,
 	// 3. Delete workflow mutable state,
 	// 4. Delete history branch.
@@ -1000,6 +1000,7 @@ func (s *ContextImpl) DeleteWorkflowExecution(
 	// Don't acquire shard lock or io semaphore if all stages that require lock are already processed.
 	if !stage.IsProcessed(
 		tasks.DeleteWorkflowExecutionStageVisibility |
+			tasks.DeleteWorkflowExecutionStageReplication |
 			tasks.DeleteWorkflowExecutionStageCurrent |
 			tasks.DeleteWorkflowExecutionStageMutableState) {

@@ -1011,11 +1012,11 @@ func (s *ContextImpl) DeleteWorkflowExecution(
 			}
 			defer s.ioSemaphoreRelease()

-			// Stage 1. Delete visibility.
-			if deleteVisibilityRecord && !stage.IsProcessed(tasks.DeleteWorkflowExecutionStageVisibility) {
-				// TODO: move to existing task generator logic
-				newTasks := map[tasks.Category][]tasks.Task{
-					tasks.CategoryVisibility: {
+			// Stage 1. Add visibility delete task and delete execution replication task.
+			if !stage.IsProcessed(tasks.DeleteWorkflowExecutionStageVisibility) {
+				newTasks := make(map[tasks.Category][]tasks.Task)
+				if deleteVisibilityRecord {
+					newTasks[tasks.CategoryVisibility] = []tasks.Task{
 						&tasks.DeleteExecutionVisibilityTask{
 							// TaskID is set by addTasks
 							WorkflowKey:                    key,
@@ -1023,25 +1024,44 @@ func (s *ContextImpl) DeleteWorkflowExecution(
 							CloseExecutionVisibilityTaskID: closeVisibilityTaskId,
 							CloseTime:                      workflowCloseTime,
 						},
-					},
+					}
 				}
-				addTasksRequest := &persistence.AddHistoryTasksRequest{
-					ShardID:     s.shardID,
-					NamespaceID: key.NamespaceID,
-					WorkflowID:  key.WorkflowID,
-					ArchetypeID: archetypeID,
-
-					Tasks: newTasks,
+				// Piggyback delete execution replication task on the same write to save a DB operation.
+				if s.config.EnableDeleteWorkflowExecutionReplication() &&
+					!stage.IsProcessed(tasks.DeleteWorkflowExecutionStageReplication) &&
+					archetypeID == chasm.WorkflowArchetypeID {
+					if nsEntry, err := s.GetNamespaceRegistry().GetNamespaceByID(
+						namespace.ID(key.NamespaceID),
+					); err == nil &&
+						nsEntry.ActiveInCluster(s.GetClusterMetadata().GetCurrentClusterName()) &&
+						nsEntry.ReplicationPolicy() == namespace.ReplicationPolicyMultiCluster {
+						newTasks[tasks.CategoryReplication] = []tasks.Task{
+							&tasks.DeleteExecutionReplicationTask{
+								WorkflowKey: key,
+								ArchetypeID: archetypeID,
+							},
+						}
+					}
 				}
-				err := s.addTasksSemaphoreAcquired(ctx, addTasksRequest)
-				if persistence.OperationPossiblySucceeded(err) {
-					engine.NotifyNewTasks(newTasks)
-				}
-				if err != nil {
-					return err
+				if len(newTasks) > 0 {
+					addTasksRequest := &persistence.AddHistoryTasksRequest{
+						ShardID:     s.shardID,
+						NamespaceID: key.NamespaceID,
+						WorkflowID:  key.WorkflowID,
+						ArchetypeID: archetypeID,
+						Tasks:       newTasks,
+					}
+					err := s.addTasksSemaphoreAcquired(ctx, addTasksRequest)
+					if persistence.OperationPossiblySucceeded(err) {
+						engine.NotifyNewTasks(newTasks)
+					}
+					if err != nil {
+						return err
+					}
 				}
 			}
 			stage.MarkProcessed(tasks.DeleteWorkflowExecutionStageVisibility)
+			stage.MarkProcessed(tasks.DeleteWorkflowExecutionStageReplication)

 			// Stage 2. Delete current workflow execution pointer.
 			if !stage.IsProcessed(tasks.DeleteWorkflowExecutionStageCurrent) {
--- a/service/history/shard/context_test.go
+++ b/service/history/shard/context_test.go
@@ -206,7 +206,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_Success() {
 	)

 	s.NoError(err)
-	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageHistory|tasks.DeleteWorkflowExecutionStageVisibility, stage)
+	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageHistory|tasks.DeleteWorkflowExecutionStageVisibility|tasks.DeleteWorkflowExecutionStageReplication, stage)
 }

 func (s *contextSuite) TestDeleteWorkflowExecution_Continue_Success() {
@@ -231,7 +231,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_Continue_Success() {
 		&stage,
 	)
 	s.NoError(err)
-	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageHistory|tasks.DeleteWorkflowExecutionStageVisibility, stage)
+	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageHistory|tasks.DeleteWorkflowExecutionStageVisibility|tasks.DeleteWorkflowExecutionStageReplication, stage)

 	s.mockExecutionManager.EXPECT().DeleteWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil)
 	s.mockExecutionManager.EXPECT().DeleteHistoryBranch(gomock.Any(), gomock.Any()).Return(nil)
@@ -246,7 +246,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_Continue_Success() {
 		&stage,
 	)
 	s.NoError(err)
-	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageHistory|tasks.DeleteWorkflowExecutionStageVisibility, stage)
+	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageHistory|tasks.DeleteWorkflowExecutionStageVisibility|tasks.DeleteWorkflowExecutionStageReplication, stage)

 	s.mockExecutionManager.EXPECT().DeleteHistoryBranch(gomock.Any(), gomock.Any()).Return(nil)
 	stage = tasks.DeleteWorkflowExecutionStageVisibility | tasks.DeleteWorkflowExecutionStageCurrent | tasks.DeleteWorkflowExecutionStageMutableState
@@ -260,7 +260,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_Continue_Success() {
 		&stage,
 	)
 	s.NoError(err)
-	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageHistory|tasks.DeleteWorkflowExecutionStageVisibility, stage)
+	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageHistory|tasks.DeleteWorkflowExecutionStageVisibility|tasks.DeleteWorkflowExecutionStageReplication, stage)
 }

 func (s *contextSuite) TestDeleteWorkflowExecution_ErrorAndContinue_Success() {
@@ -285,7 +285,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_ErrorAndContinue_Success() {
 		&stage,
 	)
 	s.Error(err)
-	s.Equal(tasks.DeleteWorkflowExecutionStageVisibility, stage)
+	s.Equal(tasks.DeleteWorkflowExecutionStageVisibility|tasks.DeleteWorkflowExecutionStageReplication, stage)

 	s.mockExecutionManager.EXPECT().DeleteCurrentWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil)
 	s.mockExecutionManager.EXPECT().DeleteWorkflowExecution(gomock.Any(), gomock.Any()).Return(errors.New("some error"))
@@ -299,7 +299,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_ErrorAndContinue_Success() {
 		&stage,
 	)
 	s.Error(err)
-	s.Equal(tasks.DeleteWorkflowExecutionStageVisibility|tasks.DeleteWorkflowExecutionStageCurrent, stage)
+	s.Equal(tasks.DeleteWorkflowExecutionStageVisibility|tasks.DeleteWorkflowExecutionStageReplication|tasks.DeleteWorkflowExecutionStageCurrent, stage)

 	s.mockExecutionManager.EXPECT().DeleteWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil)
 	s.mockExecutionManager.EXPECT().DeleteHistoryBranch(gomock.Any(), gomock.Any()).Return(errors.New("some error"))
@@ -313,7 +313,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_ErrorAndContinue_Success() {
 		&stage,
 	)
 	s.Error(err)
-	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageVisibility, stage)
+	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageVisibility|tasks.DeleteWorkflowExecutionStageReplication, stage)

 	s.mockExecutionManager.EXPECT().DeleteHistoryBranch(gomock.Any(), gomock.Any()).Return(nil)
 	err = s.mockShard.DeleteWorkflowExecution(
@@ -326,7 +326,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_ErrorAndContinue_Success() {
 		&stage,
 	)
 	s.NoError(err)
-	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageVisibility|tasks.DeleteWorkflowExecutionStageHistory, stage)
+	s.Equal(tasks.DeleteWorkflowExecutionStageCurrent|tasks.DeleteWorkflowExecutionStageMutableState|tasks.DeleteWorkflowExecutionStageVisibility|tasks.DeleteWorkflowExecutionStageHistory|tasks.DeleteWorkflowExecutionStageReplication, stage)
 }

 func (s *contextSuite) TestDeleteWorkflowExecution_DeleteVisibilityTaskNotifiction() {
@@ -366,7 +366,7 @@ func (s *contextSuite) TestDeleteWorkflowExecution_DeleteVisibilityTaskNotificti
 		&stage,
 	)
 	s.Error(err)
-	s.Equal(tasks.DeleteWorkflowExecutionStageVisibility, stage)
+	s.Equal(tasks.DeleteWorkflowExecutionStageVisibility|tasks.DeleteWorkflowExecutionStageReplication, stage)
 }

 func (s *contextSuite) TestAcquireShardOwnershipLostErrorIsNotRetried() {
--- a/service/history/tasks/delete_execution_replication_task.go
+++ b/service/history/tasks/delete_execution_replication_task.go
@@ -0,0 +1,57 @@
+package tasks
+
+import (
+	"time"
+
+	enumsspb "go.temporal.io/server/api/enums/v1"
+	"go.temporal.io/server/common/definition"
+)
+
+var _ Task = (*DeleteExecutionReplicationTask)(nil)
+var _ HasArchetypeID = (*DeleteExecutionReplicationTask)(nil)
+
+type DeleteExecutionReplicationTask struct {
+	definition.WorkflowKey
+	VisibilityTimestamp time.Time
+	TaskID              int64
+	ArchetypeID         uint32
+}
+
+func (a *DeleteExecutionReplicationTask) GetKey() Key {
+	return NewImmediateKey(a.TaskID)
+}
+
+func (a *DeleteExecutionReplicationTask) GetVersion() int64 {
+	return 0
+}
+
+func (a *DeleteExecutionReplicationTask) SetVersion(_ int64) {
+}
+
+func (a *DeleteExecutionReplicationTask) GetTaskID() int64 {
+	return a.TaskID
+}
+
+func (a *DeleteExecutionReplicationTask) SetTaskID(id int64) {
+	a.TaskID = id
+}
+
+func (a *DeleteExecutionReplicationTask) GetVisibilityTime() time.Time {
+	return a.VisibilityTimestamp
+}
+
+func (a *DeleteExecutionReplicationTask) SetVisibilityTime(timestamp time.Time) {
+	a.VisibilityTimestamp = timestamp
+}
+
+func (a *DeleteExecutionReplicationTask) GetCategory() Category {
+	return CategoryReplication
+}
+
+func (a *DeleteExecutionReplicationTask) GetType() enumsspb.TaskType {
+	return enumsspb.TASK_TYPE_REPLICATION_DELETE_EXECUTION
+}
+
+func (a *DeleteExecutionReplicationTask) GetArchetypeID() uint32 {
+	return a.ArchetypeID
+}
--- a/service/history/tasks/delete_workflow_execution_stage.go
+++ b/service/history/tasks/delete_workflow_execution_stage.go
@@ -9,6 +9,7 @@ const (

 const (
 	DeleteWorkflowExecutionStageVisibility DeleteWorkflowExecutionStage = 1 << iota
+	DeleteWorkflowExecutionStageReplication
 	DeleteWorkflowExecutionStageCurrent
 	DeleteWorkflowExecutionStageMutableState
 	DeleteWorkflowExecutionStageHistory
--- a/tests/xdc/delete_execution_replication_test.go
+++ b/tests/xdc/delete_execution_replication_test.go
@@ -0,0 +1,401 @@
+package xdc
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/suite"
+	commandpb "go.temporal.io/api/command/v1"
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/api/serviceerror"
+	taskqueuepb "go.temporal.io/api/taskqueue/v1"
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/common"
+	"go.temporal.io/server/common/config"
+	"go.temporal.io/server/common/dynamicconfig"
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/payloads"
+	"go.temporal.io/server/common/primitives"
+	"go.temporal.io/server/tests/testcore"
+	"go.uber.org/fx"
+	"google.golang.org/protobuf/types/known/durationpb"
+)
+
+type deleteExecutionReplicationTestSuite struct {
+	xdcBaseSuite
+}
+
+func TestDeleteExecutionReplicationTestSuite(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		name                    string
+		enableTransitionHistory bool
+	}{
+		{
+			name:                    "DisableTransitionHistory",
+			enableTransitionHistory: false,
+		},
+		{
+			name:                    "EnableTransitionHistory",
+			enableTransitionHistory: true,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			s := &deleteExecutionReplicationTestSuite{}
+			s.enableTransitionHistory = tc.enableTransitionHistory
+			suite.Run(t, s)
+		})
+	}
+}
+
+func (s *deleteExecutionReplicationTestSuite) SetupSuite() {
+	s.dynamicConfigOverrides = map[dynamicconfig.Key]any{
+		dynamicconfig.EnableReplicationStream.Key():                   true,
+		dynamicconfig.EnableReplicationTaskBatching.Key():             true,
+		dynamicconfig.EnableDeleteWorkflowExecutionReplication.Key():  true,
+		dynamicconfig.EnableSeparateReplicationEnableFlag.Key():       true,
+		dynamicconfig.EnableWorkflowTaskStampIncrementOnFailure.Key(): true,
+	}
+	s.logger = log.NewTestLogger()
+
+	s.setupSuite(
+		testcore.WithFxOptionsForService(primitives.AllServices,
+			fx.Decorate(
+				func(_ config.DCRedirectionPolicy) config.DCRedirectionPolicy {
+					return config.DCRedirectionPolicy{Policy: "noop"}
+				},
+			),
+		),
+	)
+}
+
+func (s *deleteExecutionReplicationTestSuite) TearDownSuite() {
+	s.tearDownSuite()
+}
+
+func (s *deleteExecutionReplicationTestSuite) SetupTest() {
+	s.setupTest()
+}
+
+func (s *deleteExecutionReplicationTestSuite) TestDeleteClosedWorkflow_ReplicatedToPassiveCluster() {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute*2)
+	defer cancel()
+
+	// Create a global namespace on both clusters.
+	ns := s.createGlobalNamespace()
+	nsResp, err := s.clusters[0].FrontendClient().DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{
+		Namespace: ns,
+	})
+	s.Require().NoError(err)
+	nsID := nsResp.GetNamespaceInfo().GetId()
+
+	workflowID := "test-delete-replication-" + uuid.NewString()
+	taskQueue := "test-delete-tq-" + uuid.NewString()
+	sourceClient := s.clusters[0].FrontendClient()
+
+	// Start a workflow on the active cluster.
+	startResp, err := sourceClient.StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
+		RequestId:           uuid.NewString(),
+		Namespace:           ns,
+		WorkflowId:          workflowID,
+		WorkflowType:        &commonpb.WorkflowType{Name: "test-wf-type"},
+		TaskQueue:           &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
+		WorkflowRunTimeout:  durationpb.New(time.Minute),
+		WorkflowTaskTimeout: durationpb.New(10 * time.Second),
+	})
+	s.Require().NoError(err)
+	runID := startResp.GetRunId()
+
+	// Complete the workflow.
+	wtHandler := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
+		return []*commandpb.Command{{
+			CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
+			Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
+				CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
+					Result: payloads.EncodeString("done"),
+				},
+			},
+		}}, nil
+	}
+	//nolint:staticcheck // TODO: replace with taskpoller.TaskPoller
+	poller := &testcore.TaskPoller{
+		Client:              sourceClient,
+		Namespace:           ns,
+		TaskQueue:           &taskqueuepb.TaskQueue{Name: taskQueue},
+		Identity:            "worker",
+		WorkflowTaskHandler: wtHandler,
+		Logger:              s.logger,
+		T:                   s.T(),
+	}
+	_, err = poller.PollAndProcessWorkflowTask()
+	s.Require().NoError(err)
+
+	// Wait for workflow to be completed on active cluster.
+	s.Eventually(func() bool {
+		resp, err := sourceClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: ns,
+			Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
+		})
+		if err != nil {
+			return false
+		}
+		return resp.GetWorkflowExecutionInfo().GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
+	}, time.Second*10, time.Second)
+
+	// Wait for the workflow to be replicated to the passive cluster.
+	targetClient := s.clusters[1].FrontendClient()
+	s.Eventually(func() bool {
+		resp, err := targetClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: ns,
+			Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
+		})
+		if err != nil {
+			return false
+		}
+		return resp.GetWorkflowExecutionInfo().GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
+	}, replicationWaitTime, replicationCheckInterval, "Workflow should be replicated to passive cluster")
+
+	// Delete the workflow on the active cluster.
+	_, err = sourceClient.DeleteWorkflowExecution(ctx, &workflowservice.DeleteWorkflowExecutionRequest{
+		Namespace: ns,
+		WorkflowExecution: &commonpb.WorkflowExecution{
+			WorkflowId: workflowID,
+			RunId:      runID,
+		},
+	})
+	s.Require().NoError(err)
+
+	// Verify the workflow is deleted on the active cluster.
+	s.Eventually(func() bool {
+		_, err := sourceClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: ns,
+			Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
+		})
+		if err == nil {
+			return false
+		}
+		var notFound *serviceerror.NotFound
+		return errors.As(err, &notFound)
+	}, time.Second*10, time.Second, "Workflow should be deleted on active cluster")
+
+	// Verify the workflow mutable state is deleted on the passive cluster via replication.
+	s.Eventually(func() bool {
+		_, err := s.clusters[1].HistoryClient().DescribeMutableState(ctx, &historyservice.DescribeMutableStateRequest{
+			NamespaceId: nsID,
+			Execution:   &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
+			ArchetypeId: chasm.WorkflowArchetypeID,
+		})
+		if err == nil {
+			return false
+		}
+		var notFound *serviceerror.NotFound
+		return errors.As(err, &notFound)
+	}, time.Second*30, replicationCheckInterval, "Workflow mutable state should be deleted on passive cluster via replication")
+}
+
+func (s *deleteExecutionReplicationTestSuite) TestDeleteRunningWorkflow_ReplicatedToPassiveCluster() {
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+
+	ns := s.createGlobalNamespace()
+	nsResp, err := s.clusters[0].FrontendClient().DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{
+		Namespace: ns,
+	})
+	s.Require().NoError(err)
+	nsID := nsResp.GetNamespaceInfo().GetId()
+
+	workflowID := "test-delete-running-" + uuid.NewString()
+	taskQueue := "test-delete-running-tq-" + uuid.NewString()
+	sourceClient := s.clusters[0].FrontendClient()
+
+	// Start a workflow on the active cluster (don't complete it — leave it running).
+	startResp, err := sourceClient.StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
+		RequestId:           uuid.NewString(),
+		Namespace:           ns,
+		WorkflowId:          workflowID,
+		WorkflowType:        &commonpb.WorkflowType{Name: "test-wf-type"},
+		TaskQueue:           &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
+		WorkflowRunTimeout:  durationpb.New(time.Minute),
+		WorkflowTaskTimeout: durationpb.New(10 * time.Second),
+	})
+	s.Require().NoError(err)
+	runID := startResp.GetRunId()
+
+	// Wait for the workflow to be replicated to the passive cluster.
+	s.Eventually(func() bool {
+		_, err := s.clusters[1].FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: ns,
+			Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
+		})
+		return err == nil
+	}, replicationWaitTime, replicationCheckInterval, "Workflow should be replicated to passive cluster")
+
+	// Delete the running workflow on the active cluster.
+	// This will terminate it first (deleteAfterTerminate=true), then delete.
+	_, err = sourceClient.DeleteWorkflowExecution(ctx, &workflowservice.DeleteWorkflowExecutionRequest{
+		Namespace: ns,
+		WorkflowExecution: &commonpb.WorkflowExecution{
+			WorkflowId: workflowID,
+			RunId:      runID,
+		},
+	})
+	s.Require().NoError(err)
+
+	// Verify the workflow is deleted on the active cluster.
+	s.Eventually(func() bool {
+		_, err := sourceClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: ns,
+			Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
+		})
+		if err == nil {
+			return false
+		}
+		var notFound *serviceerror.NotFound
+		return errors.As(err, &notFound)
+	}, time.Second*10, time.Second, "Workflow should be deleted on active cluster")
+
+	// Verify the workflow mutable state is deleted on the passive cluster via replication.
+	s.Eventually(func() bool {
+		_, err := s.clusters[1].HistoryClient().DescribeMutableState(ctx, &historyservice.DescribeMutableStateRequest{
+			NamespaceId: nsID,
+			Execution:   &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
+			ArchetypeId: chasm.WorkflowArchetypeID,
+		})
+		if err == nil {
+			return false
+		}
+		var notFound *serviceerror.NotFound
+		return errors.As(err, &notFound)
+	}, time.Second*30, replicationCheckInterval, "Workflow mutable state should be deleted on passive cluster via replication")
+}
+
+func (s *deleteExecutionReplicationTestSuite) TestDeleteWorkflow_NotReplicatedWhenFeatureFlagDisabled() {
+	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
+	defer cancel()
+
+	// Disable the feature flag on both clusters.
+	cleanup0 := s.clusters[0].OverrideDynamicConfig(s.T(), dynamicconfig.EnableDeleteWorkflowExecutionReplication, false)
+	defer cleanup0()
+	cleanup1 := s.clusters[1].OverrideDynamicConfig(s.T(), dynamicconfig.EnableDeleteWorkflowExecutionReplication, false)
+	defer cleanup1()
+
+	ns := s.createGlobalNamespace()
+
+	workflowID := "test-delete-no-repl-" + uuid.NewString()
+	taskQueue := "test-delete-no-repl-tq-" + uuid.NewString()
+	sourceClient := s.clusters[0].FrontendClient()
+
+	// Start and complete a workflow.
+	startResp, err := sourceClient.StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
+		RequestId:           uuid.NewString(),
+		Namespace:           ns,
+		WorkflowId:          workflowID,
+		WorkflowType:        &commonpb.WorkflowType{Name: "test-wf-type"},
+		TaskQueue:           &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
+		WorkflowRunTimeout:  durationpb.New(time.Minute),
+		WorkflowTaskTimeout: durationpb.New(10 * time.Second),
+	})
+	s.Require().NoError(err)
+	runID := startResp.GetRunId()
+
+	wtHandler := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
+		return []*commandpb.Command{{
+			CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
+			Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
+				CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
+					Result: payloads.EncodeString("done"),
+				},
+			},
+		}}, nil
+	}
+	//nolint:staticcheck // TODO: replace with taskpoller.TaskPoller
+	poller := &testcore.TaskPoller{
+		Client:              sourceClient,
+		Namespace:           ns,
+		TaskQueue:           &taskqueuepb.TaskQueue{Name: taskQueue},
+		Identity:            "worker",
+		WorkflowTaskHandler: wtHandler,
+		Logger:              s.logger,
+		T:                   s.T(),
+	}
+	_, err = poller.PollAndProcessWorkflowTask()
+	s.Require().NoError(err)
+
+	// Wait for replication to passive.
+	targetClient := s.clusters[1].FrontendClient()
+	s.Eventually(func() bool {
+		resp, err := targetClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: ns,
+			Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
+		})
+		if err != nil {
+			return false
+		}
+		return resp.GetWorkflowExecutionInfo().GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
+	}, replicationWaitTime, replicationCheckInterval)
+
+	// Delete on active cluster.
+	_, err = sourceClient.DeleteWorkflowExecution(ctx, &workflowservice.DeleteWorkflowExecutionRequest{
+		Namespace: ns,
+		WorkflowExecution: &commonpb.WorkflowExecution{
+			WorkflowId: workflowID,
+			RunId:      runID,
+		},
+	})
+	s.Require().NoError(err)
+
+	// Wait for deletion on active.
+	s.Eventually(func() bool {
+		_, err := sourceClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: ns,
+			Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
+		})
+		var notFound *serviceerror.NotFound
+		return errors.As(err, &notFound)
+	}, time.Second*10, time.Second)
+
+	// Workflow should still exist on the passive cluster (no replication of deletion).
+	//nolint:forbidigo // need to wait to confirm deletion did NOT replicate
+	time.Sleep(5 * time.Second)
+	_, err = targetClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+		Namespace: ns,
+		Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID},
+	})
+	s.NoError(err, "Workflow should still exist on passive cluster when feature flag is disabled")
+}
+
+func (s *deleteExecutionReplicationTestSuite) createGlobalNamespace() string {
+	ctx := testcore.NewContext()
+	ns := "test-delete-ns-" + common.GenerateRandomString(5)
+	_, err := s.clusters[0].FrontendClient().RegisterNamespace(ctx, &workflowservice.RegisterNamespaceRequest{
+		Namespace:                        ns,
+		IsGlobalNamespace:                true,
+		Clusters:                         s.clusterReplicationConfig(),
+		ActiveClusterName:                s.clusters[0].ClusterName(),
+		WorkflowExecutionRetentionPeriod: durationpb.New(24 * time.Hour),
+	})
+	s.Require().NoError(err)
+
+	// Wait for namespace to be available on both clusters.
+	s.Eventually(func() bool {
+		_, err := s.clusters[0].FrontendClient().DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{
+			Namespace: ns,
+		})
+		return err == nil
+	}, namespaceCacheWaitTime, namespaceCacheCheckInterval)
+
+	s.Eventually(func() bool {
+		_, err := s.clusters[1].FrontendClient().DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{
+			Namespace: ns,
+		})
+		return err == nil
+	}, namespaceCacheWaitTime, namespaceCacheCheckInterval)
+
+	return ns
+}