mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-02 12:22:18 +00:00
jiechenzandGitHub
32c0915b24
## 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
61 lines
2.1 KiB
Go
61 lines
2.1 KiB
Go
package replication
|
|
|
|
import (
|
|
enumsspb "go.temporal.io/server/api/enums/v1"
|
|
replicationspb "go.temporal.io/server/api/replication/v1"
|
|
"go.temporal.io/server/common/metrics"
|
|
)
|
|
|
|
func TaskOperationTag(
|
|
replicationTask *replicationspb.ReplicationTask,
|
|
) string {
|
|
if replicationTask == nil {
|
|
return "__unknown__"
|
|
}
|
|
switch replicationTask.GetTaskType() {
|
|
case enumsspb.REPLICATION_TASK_TYPE_SYNC_SHARD_STATUS_TASK: // TODO to be deprecated
|
|
return "__unknown__"
|
|
case enumsspb.REPLICATION_TASK_TYPE_HISTORY_METADATA_TASK: // TODO to be deprecated
|
|
return "__unknown__"
|
|
case enumsspb.REPLICATION_TASK_TYPE_SYNC_ACTIVITY_TASK:
|
|
return metrics.SyncActivityTaskScope
|
|
case enumsspb.REPLICATION_TASK_TYPE_SYNC_WORKFLOW_STATE_TASK:
|
|
return metrics.SyncWorkflowStateTaskScope
|
|
case enumsspb.REPLICATION_TASK_TYPE_HISTORY_V2_TASK:
|
|
return metrics.HistoryReplicationTaskScope
|
|
case enumsspb.REPLICATION_TASK_TYPE_SYNC_HSM_TASK:
|
|
return metrics.SyncHSMTaskScope
|
|
case enumsspb.REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK:
|
|
return metrics.SyncVersionedTransitionTaskScope
|
|
case enumsspb.REPLICATION_TASK_TYPE_BACKFILL_HISTORY_TASK:
|
|
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
|
|
}
|
|
}
|
|
|
|
func TaskOperationTagFromTask(
|
|
taskType enumsspb.TaskType,
|
|
) string {
|
|
switch taskType {
|
|
case enumsspb.TASK_TYPE_REPLICATION_SYNC_HSM:
|
|
return metrics.SyncHSMTaskScope
|
|
case enumsspb.TASK_TYPE_REPLICATION_SYNC_VERSIONED_TRANSITION:
|
|
return metrics.SyncVersionedTransitionTaskScope
|
|
case enumsspb.TASK_TYPE_REPLICATION_SYNC_ACTIVITY:
|
|
return metrics.SyncActivityTaskScope
|
|
case enumsspb.TASK_TYPE_REPLICATION_SYNC_WORKFLOW_STATE:
|
|
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
|
|
}
|
|
}
|