Handling of duplicate StartWorkflowExecution event processing
Replicator could potentially see duplicate StartWorkflowExecution
replication events.  Adding logic to replicator to dedupe such events.
@@ -78,21 +78,33 @@ func (r *historyReplicator) ApplyEvents(request *h.ReplicateEventsRequest) (retE
execution := *request.WorkflowExecution
- var context *workflowExecutionContext
+ context, release, err := r.historyCache.getOrCreateWorkflowExecution(domainID, execution)
+ if err != nil {
+ return err
+ }
+ defer func() { release(retError) }()
+
var msBuilder *mutableStateBuilder
firstEvent := request.History.Events[0]
switch firstEvent.GetEventType() {
case shared.EventTypeWorkflowExecutionStarted:
- msBuilder = newMutableStateBuilderWithReplicationState(r.shard.GetConfig(), r.logger, request.GetVersion())
+ msBuilder, err = context.loadWorkflowExecution()
+ if err == nil {
+ // Workflow execution already exist, looks like a duplicate start event, it is safe to ignore it
+ r.logger.Infof("Dropping stale replication task for start event. WorkflowID: %v, RunID: %v, Version: %v",
+ execution.GetWorkflowId(), execution.GetRunId(), request.GetVersion())
+ return nil
+ }
- default:
- var release releaseWorkflowExecutionFunc
- context, release, err = r.historyCache.getOrCreateWorkflowExecution(domainID, execution)
- if err != nil {
+ // GetWorkflowExecution failed with some transient error. Return err so we can retry the task later
+ if _, ok := err.(*shared.EntityNotExistsError); !ok {
return err
}
- defer func() { release(retError) }()
+ // WorkflowExecution does not exist, lets proceed with processing of the task
+ msBuilder = newMutableStateBuilderWithReplicationState(r.shard.GetConfig(), r.logger, request.GetVersion())
+
+ default:
msBuilder, err = context.loadWorkflowExecution()
if err != nil {
return err
@@ -305,34 +317,35 @@ func (r *historyReplicator) ApplyReplicationTask(context *workflowExecutionConte
// The failover version checking && overwriting should be performed here: #675
// TODO
- workflowExistsErrHandler := func(err *persistence.WorkflowExecutionAlreadyStartedError) error {
- // set the prev run ID for database conditional update
- prevRunID := err.RunID
- prevState := err.State
- if prevState != persistence.WorkflowStateCompleted {
+ // try to create the workflow execution
+ isBrandNew := true
+ _, err = createWorkflow(isBrandNew, "")
+ // if err still non nil, see if retry
+ if errExist, ok := err.(*persistence.WorkflowExecutionAlreadyStartedError); ok {
+ prevRunID := errExist.RunID
+ prevState := errExist.State
+
+ // Check for duplicate processing of StartWorkflowExecution replication task
if prevRunID == execution.GetRunId() {
- // this is a duplicate execution of the start execution event
- // or this is an execution of the duplicate start execution event
+ r.logger.Infof("Dropping stale replication task for start event. WorkflowID: %v, RunID: %v, Version: %v",
+ execution.GetWorkflowId(), execution.GetRunId(), request.GetVersion())
return nil
}
+
+ // Some other workflow is running, for now let's keep on retrying this replication event by an error
+ // to wait for current run to finish so this event could be applied.
+ // TODO: We also need to deal with conflict resolution when workflow with same ID is started on 2 different
+ // clusters.
+ if prevState != persistence.WorkflowStateCompleted {
return err
}
+
// if the existing workflow is completed, ignore the worklow ID reuse policy
// since the policy should be applied by the active cluster,
// standby cluster should apply this event without question.
- return nil
- }
-
- // try to create the workflow execution
- isBrandNew := true
- _, err = createWorkflow(isBrandNew, "")
- // if err still non nil, see if retry
- if errExist, ok := err.(*persistence.WorkflowExecutionAlreadyStartedError); ok {
- if err = workflowExistsErrHandler(errExist); err == nil {
isBrandNew = false
_, err = createWorkflow(isBrandNew, errExist.RunID)
}
- }
default:
// Generate a transaction ID for appending events to history
