Support unpinned workflows #6887
Open
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…g-behaviors # Conflicts: # api/deployment/v1/message.go-helpers.pb.go # api/deployment/v1/message.pb.go # common/worker_versioning/worker_versioning.go # proto/internal/temporal/server/api/deployment/v1/message.proto
…g-behaviors # Conflicts: # api/persistence/v1/executions.pb.go # service/matching/task_queue_partition_manager.go
…shahab/transfer-versioning-info # Conflicts: # service/history/workflow/mutable_state_impl.go
…er-versioning-info # Conflicts: # api/deployment/v1/message.pb.go # api/historyservice/v1/request_response.pb.go # api/persistence/v1/tasks.pb.go # api/taskqueue/v1/message.pb.go # proto/internal/temporal/server/api/deployment/v1/message.proto # proto/internal/temporal/server/api/historyservice/v1/request_response.proto
…haviors # Conflicts: # common/worker_versioning/worker_versioning.go
…er-versioning-info
…g-behaviors # Conflicts: # common/worker_versioning/worker_versioning.go # service/history/workflow/util.go
ShahabT
changed the base branch from
versioning-3
to
shahab/routing-behaviors
alexshtin
reviewed
Nov 26, 2024
| func Deployment(deployment *deployment.Deployment) ZapTag { | ||
| val := "UNVERSIONED" | ||
| if deployment != nil { | ||
| val = deployment.SeriesName + ":" + deployment.GetBuildId() |
There was a problem hiding this comment.
I would use GetXXX() format everywhere for consistency. Also all these log tag thing is aiming to 0 allocations and 0 overhead if tag is not used but I guess there is no good way to make it work here.
There was a problem hiding this comment.
I would make deployment a string here and call DeploymentToString before passing deployment to this func. Let's make tags trivial simple.
| @@ -110,7 +110,7 @@ message VersioningData { | |||
| message DeploymentData { | |||
| // Set of deployments that this task queue belongs to. | |||
| // Current deployment is defined implicitly as the deployment with the most recent last_became_current_time. | |||
There was a problem hiding this comment.
Can you specify TaskQueueData.last_became_current_time?
| @@ -110,7 +110,7 @@ message VersioningData { | |||
| message DeploymentData { | |||
| // Set of deployments that this task queue belongs to. | |||
| // Current deployment is defined implicitly as the deployment with the most recent last_became_current_time. | |||
| repeated Deployment deployment = 1; | |||
| repeated Deployment deployments = 1; | |||
|
|
|||
| message Deployment { | |||
There was a problem hiding this comment.
This nested Deployment is a bad sign. DeploymentDataItem?
| } | ||
| wfDeployment := mutableState.GetEffectiveDeployment() | ||
| if !directiveDeployment.Equal(wfDeployment) { | ||
| // This must be a task scheduled before the workflow transitions to the current |
There was a problem hiding this comment.
Same way as we refer "workflow task" as WFT everywhere in comments, I would suggest to refer "activity task" as AT, not just "task".
| return nil, serviceerrors.NewObsoleteMatchingTask("wrong task queue type") | ||
| } | ||
|
|
||
| dispatchDeployment := worker_versioning.DeploymentFromCapabilities(req.PollRequest.WorkerVersionCapabilities) |
There was a problem hiding this comment.
Suggested change
| dispatchDeployment := worker_versioning.DeploymentFromCapabilities(req.PollRequest.WorkerVersionCapabilities) | |
| pollerDeployment := worker_versioning.DeploymentFromCapabilities(req.PollRequest.WorkerVersionCapabilities) |
| } | ||
|
|
||
| dispatchDeployment := worker_versioning.DeploymentFromCapabilities(req.PollRequest.WorkerVersionCapabilities) | ||
| directiveDeployment := req.GetDirectiveDeployment() |
There was a problem hiding this comment.
Suggested change
| directiveDeployment := req.GetDirectiveDeployment() | |
| scheduledDeployment := req.GetScheduledDeployment() |
-- deployment which history wanted when it scheduled a WFT.
| if errors.Is(err, workflow.ErrPinnedWorkflowCannotTransition) { | ||
| // This must be a task from a time that the workflow was unpinned, but it's | ||
| // now pinned so can't transition. Matching can drop the task safely. | ||
| return nil, serviceerrors.NewObsoleteMatchingTask("workflow is not eligible for transition") |
There was a problem hiding this comment.
Suggested change
| return nil, serviceerrors.NewObsoleteMatchingTask("workflow is not eligible for transition") | |
| return nil, serviceerrors.NewObsoleteMatchingTask(workflow.ErrPinnedWorkflowCannotTransition.Error()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What changed?
Add Matching and History changes to properly route unpinned workflow tasks.
Why?
How did you test it?
Tests will come in separate PR.
Potential risks
Documentation
Is hotfix candidate?