Special Handling for new workflow replication #7561
Conversation
| } | ||
| } | ||
|
|
||
| wfCtx, releaseFn, err := r.workflowCache.GetOrCreateWorkflowExecution( |
There was a problem hiding this comment.
Could use current workflow lock.
There was a problem hiding this comment.
GetOrCreateWorkflowExecution does not return the workflow context which is necessary parameter for ndc createWorkflow.
There was a problem hiding this comment.
I see. Looks like normal start workflow (manually creates the workflowContext object) and replication start workflow follows a different approach.
|
|
||
| if mutation != nil && mutation.ExclusiveStartVersionedTransition.TransitionCount == 0 { | ||
| // this is the first replication task for this workflow | ||
| // TODO: Handle reset case to reduce the amount of history events write |
| err = r.taskRefresher.Refresh(ctx, localMutableState) | ||
|
|
||
| if err != nil { | ||
| println(err.Error()) |
| //nolint:revive // cognitive complexity 35 (> max enabled 25) | ||
| func (r *WorkflowStateReplicatorImpl) handleFirstReplicationTask( | ||
| ctx context.Context, | ||
| versionedTransition *replicationspb.VersionedTransitionArtifact, |
| return err | ||
| } | ||
|
|
||
| localMutableState.SetHistoryBuilder(historybuilder.NewImmutable(historyEventBatchs...)) |
There was a problem hiding this comment.
will this update the HistorySize in ExecutionStats?
There was a problem hiding this comment.
Yes, it is handled here:
| SyncWorkflowStateMutationAttributes: &replicationspb.SyncWorkflowStateMutationAttributes{ | ||
| StateMutation: mutation, | ||
| ExclusiveStartVersionedTransition: &persistencespb.VersionedTransition{ | ||
| NamespaceFailoverVersion: taskVersionedTransition.NamespaceFailoverVersion, |
There was a problem hiding this comment.
so this is for making sure workflow.TransitionHistoryStalenessCheck(localTransitionHistory, mutation.ExclusiveStartVersionedTransition) != nil check can pass when applying mutation?
There was a problem hiding this comment.
yes. so we don't need special handling for that.
| } | ||
| } | ||
|
|
||
| wfCtx, releaseFn, err := r.workflowCache.GetOrCreateWorkflowExecution( |
There was a problem hiding this comment.
I see. Looks like normal start workflow (manually creates the workflowContext object) and replication start workflow follows a different approach.
What changed?
Add special handling for new workflow replication
Why?
To reduce passive side loadmutablestate not found attempt
How did you test it?
unit test
Potential risks
no risk
Documentation
n/a
Is hotfix candidate?
no