diff --git a/go.mod b/go.mod index b4c1c938c84..dd12769c8bd 100644 --- a/go.mod +++ b/go.mod @@ -58,7 +58,7 @@ require ( go.opentelemetry.io/otel/sdk v1.34.0 go.opentelemetry.io/otel/sdk/metric v1.34.0 go.opentelemetry.io/otel/trace v1.34.0 - go.temporal.io/api v1.58.1-0.20251126231839-2fcd2247e106 + go.temporal.io/api v1.58.1-0.20251128181858-703071215042 go.temporal.io/sdk v1.35.0 go.uber.org/fx v1.24.0 go.uber.org/mock v0.6.0 diff --git a/go.sum b/go.sum index b6cadbeeb2c..ea33209e437 100644 --- a/go.sum +++ b/go.sum @@ -390,8 +390,8 @@ go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE= go.opentelemetry.io/proto/otlp v1.5.0 h1:xJvq7gMzB31/d406fB8U5CBdyQGw4P399D1aQWU/3i4= go.opentelemetry.io/proto/otlp v1.5.0/go.mod h1:keN8WnHxOy8PG0rQZjJJ5A2ebUoafqWp0eVQ4yIXvJ4= -go.temporal.io/api v1.58.1-0.20251126231839-2fcd2247e106 h1:V2H8rfBDapmWpIsNDWZLOS95WIIWAgPXnG7gpNrWO5Y= -go.temporal.io/api v1.58.1-0.20251126231839-2fcd2247e106/go.mod h1:iaxoP/9OXMJcQkETTECfwYq4cw/bj4nwov8b3ZLVnXM= +go.temporal.io/api v1.58.1-0.20251128181858-703071215042 h1:44+nPe+rGhYUwA1oDi46rkXEYEVfoAxOmb0myvTm4Es= +go.temporal.io/api v1.58.1-0.20251128181858-703071215042/go.mod h1:iaxoP/9OXMJcQkETTECfwYq4cw/bj4nwov8b3ZLVnXM= go.temporal.io/sdk v1.35.0 h1:lRNAQ5As9rLgYa7HBvnmKyzxLcdElTuoFJ0FXM/AsLQ= go.temporal.io/sdk v1.35.0/go.mod h1:1q5MuLc2MEJ4lneZTHJzpVebW2oZnyxoIOWX3oFVebw= go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= diff --git a/service/frontend/workflow_handler.go b/service/frontend/workflow_handler.go index 041a8e91afd..d8904cbc772 100644 --- a/service/frontend/workflow_handler.go +++ b/service/frontend/workflow_handler.go @@ -5950,6 +5950,9 @@ func (wh *WorkflowHandler) UpdateWorkflowExecutionOptions( if err != nil { return nil, serviceerror.NewInvalidArgumentf("error parsing UpdateMask: %s", err.Error()) } + if err := priorities.Validate(opts.GetPriority()); err != nil { + return nil, err + } namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace())) if err != nil { @@ -5987,6 +5990,9 @@ func (wh *WorkflowHandler) UpdateActivityOptions( if request.GetActivity() == nil { return nil, errActivityIDOrTypeNotSet } + if err := priorities.Validate(request.GetActivityOptions().GetPriority()); err != nil { + return nil, err + } namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace())) if err != nil { diff --git a/service/frontend/workflow_handler_test.go b/service/frontend/workflow_handler_test.go index 28097f59e51..061ecda3e28 100644 --- a/service/frontend/workflow_handler_test.go +++ b/service/frontend/workflow_handler_test.go @@ -14,6 +14,7 @@ import ( "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" + activitypb "go.temporal.io/api/activity/v1" batchpb "go.temporal.io/api/batch/v1" commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" @@ -914,6 +915,31 @@ func (s *WorkflowHandlerSuite) TestStartWorkflowExecution_Failed_InvalidAggregat s.ErrorContains(err, "cannot attach more than 10 links per request, got 11") } +func (s *WorkflowHandlerSuite) TestStartWorkflowExecution_Priority() { + config := s.newConfig() + wh := s.getWorkflowHandler(config) + + s.mockSearchAttributesMapperProvider.EXPECT().GetMapper(gomock.Any()).Return(nil, nil).AnyTimes() + + request := &workflowservice.StartWorkflowExecutionRequest{ + Namespace: s.testNamespace.String(), + WorkflowId: "workflow-id", + WorkflowType: &commonpb.WorkflowType{ + Name: "workflow-type", + }, + TaskQueue: &taskqueuepb.TaskQueue{ + Name: "task-queue", + }, + Priority: &commonpb.Priority{PriorityKey: -1}, + } + + _, err := wh.StartWorkflowExecution(context.Background(), request) + var invalidArg *serviceerror.InvalidArgument + s.ErrorAs(err, &invalidArg) + s.ErrorContains(err, "priority key can't be negative") + // NOTE: only testing a single validation scenario here; the priority validation has its own unit tests +} + func (s *WorkflowHandlerSuite) TestSignalWithStartWorkflowExecution_InvalidWorkflowIdConflictPolicy() { config := s.newConfig() wh := s.getWorkflowHandler(config) @@ -1011,6 +1037,32 @@ func (s *WorkflowHandlerSuite) TestSignalWithStartWorkflowExecution_Failed_Inval s.ErrorContains(err, "link exceeds allowed size of 4000") } +func (s *WorkflowHandlerSuite) TestSignalWithStartWorkflowExecution_Priority() { + config := s.newConfig() + wh := s.getWorkflowHandler(config) + + s.mockSearchAttributesMapperProvider.EXPECT().GetMapper(gomock.Any()).Return(nil, nil).AnyTimes() + + request := &workflowservice.SignalWithStartWorkflowExecutionRequest{ + Namespace: s.testNamespace.String(), + WorkflowId: "workflow-id", + WorkflowType: &commonpb.WorkflowType{ + Name: "workflow-type", + }, + TaskQueue: &taskqueuepb.TaskQueue{ + Name: "task-queue", + }, + SignalName: "signal-name", + Priority: &commonpb.Priority{PriorityKey: -1}, + } + + _, err := wh.SignalWithStartWorkflowExecution(context.Background(), request) + var invalidArg *serviceerror.InvalidArgument + s.ErrorAs(err, &invalidArg) + s.ErrorContains(err, "priority key can't be negative") + // NOTE: only testing a single validation scenario here; the priority validation has its own unit tests +} + func (s *WorkflowHandlerSuite) TestSignalWorkflowExecution_Failed_InvalidLinks() { s.mockSearchAttributesMapperProvider.EXPECT().GetMapper(gomock.Any()).AnyTimes().Return(nil, nil) config := s.newConfig() @@ -3958,3 +4010,52 @@ func (s *WorkflowHandlerSuite) TestUpdateTaskQueueConfig_Validation() { s.NotNil(resp) }) } + +func (s *WorkflowHandlerSuite) TestUpdateWorkflowExecutionOptions_Priority() { + config := s.newConfig() + wh := s.getWorkflowHandler(config) + + request := &workflowservice.UpdateWorkflowExecutionOptionsRequest{ + Namespace: s.testNamespace.String(), + WorkflowExecution: &commonpb.WorkflowExecution{ + WorkflowId: "workflow-id", + RunId: "run-id", + }, + WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{ + Priority: &commonpb.Priority{PriorityKey: -1}, + }, + UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"priority"}}, + } + + _, err := wh.UpdateWorkflowExecutionOptions(context.Background(), request) + var invalidArg *serviceerror.InvalidArgument + s.ErrorAs(err, &invalidArg) + s.ErrorContains(err, "priority key can't be negative") + // NOTE: only testing a single validation scenario here; the priority validation has its own unit tests +} + +func (s *WorkflowHandlerSuite) TestUpdateActivityOptions_Priority() { + config := s.newConfig() + wh := s.getWorkflowHandler(config) + + request := &workflowservice.UpdateActivityOptionsRequest{ + Namespace: s.testNamespace.String(), + Execution: &commonpb.WorkflowExecution{ + WorkflowId: "workflow-id", + RunId: "run-id", + }, + Activity: &workflowservice.UpdateActivityOptionsRequest_Id{ + Id: "activity-id", + }, + ActivityOptions: &activitypb.ActivityOptions{ + Priority: &commonpb.Priority{PriorityKey: -1}, + }, + UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"priority"}}, + } + + _, err := wh.UpdateActivityOptions(context.Background(), request) + var invalidArg *serviceerror.InvalidArgument + s.ErrorAs(err, &invalidArg) + s.ErrorContains(err, "priority key can't be negative") + // NOTE: only testing a single validation scenario here; the priority validation has its own unit tests +} diff --git a/service/history/api/recordactivitytaskstarted/api.go b/service/history/api/recordactivitytaskstarted/api.go index 0c68a219994..4a8f083c721 100644 --- a/service/history/api/recordactivitytaskstarted/api.go +++ b/service/history/api/recordactivitytaskstarted/api.go @@ -165,12 +165,11 @@ func recordActivityTaskStarted( } if ai.Stamp != request.Stamp { - // activity has changes before task is started. - // ErrActivityStampMismatch is the error to indicate that requested activity has mismatched stamp + // This happens when the workflow task was rescheduled. errorMessage := fmt.Sprintf( - "Activity task with this stamp not found. Id: %s,: type: %s, current stamp: %d", + "Activity task rejected; stamp has changed. Id: %s,: type: %s, current stamp: %d", ai.ActivityId, ai.ActivityType.Name, ai.Stamp) - return nil, rejectCodeUndefined, serviceerror.NewNotFound(errorMessage) + return nil, rejectCodeUndefined, serviceerrors.NewObsoleteMatchingTask(errorMessage) } wfBehavior := mutableState.GetEffectiveVersioningBehavior() diff --git a/service/history/api/startworkflow/api.go b/service/history/api/startworkflow/api.go index 6b4bd314de6..613846794c3 100644 --- a/service/history/api/startworkflow/api.go +++ b/service/history/api/startworkflow/api.go @@ -670,7 +670,8 @@ func (s *Starter) handleUseExistingWorkflowOnConflictOptions( requestID, completionCallbacks, links, - "", + "", // identity + nil, // priority ) return api.UpdateWorkflowWithoutWorkflowTask, err }, diff --git a/service/history/api/updateactivityoptions/api.go b/service/history/api/updateactivityoptions/api.go index 7b1014c2526..1d92d64934c 100644 --- a/service/history/api/updateactivityoptions/api.go +++ b/service/history/api/updateactivityoptions/api.go @@ -13,6 +13,7 @@ import ( "go.temporal.io/api/workflowservice/v1" "go.temporal.io/server/api/historyservice/v1" persistencespb "go.temporal.io/server/api/persistence/v1" + "go.temporal.io/server/common" "go.temporal.io/server/common/definition" "go.temporal.io/server/common/namespace" "go.temporal.io/server/common/util" @@ -148,6 +149,7 @@ func processActivityOptionsUpdate( ScheduleToStartTimeout: ai.ScheduleToStartTimeout, StartToCloseTimeout: ai.StartToCloseTimeout, HeartbeatTimeout: ai.HeartbeatTimeout, + Priority: common.CloneProto(ai.Priority), RetryPolicy: &commonpb.RetryPolicy{ BackoffCoefficient: ai.RetryBackoffCoefficient, InitialInterval: ai.RetryInitialInterval, @@ -202,10 +204,48 @@ func mergeActivityOptions( mergeInto.HeartbeatTimeout = mergeFrom.HeartbeatTimeout } + if _, ok := updateFields["priority"]; ok { + mergeInto.Priority = mergeFrom.Priority + } + + if _, ok := updateFields["priority.priorityKey"]; ok { + if mergeFrom.Priority == nil { + return serviceerror.NewInvalidArgument("Priority is not provided") + } + if mergeInto.Priority == nil { + mergeInto.Priority = &commonpb.Priority{} + } + mergeInto.Priority.PriorityKey = mergeFrom.Priority.PriorityKey + } + + if _, ok := updateFields["priority.fairnessKey"]; ok { + if mergeFrom.Priority == nil { + return serviceerror.NewInvalidArgument("Priority is not provided") + } + if mergeInto.Priority == nil { + mergeInto.Priority = &commonpb.Priority{} + } + mergeInto.Priority.FairnessKey = mergeFrom.Priority.FairnessKey + } + + if _, ok := updateFields["priority.fairnessWeight"]; ok { + if mergeFrom.Priority == nil { + return serviceerror.NewInvalidArgument("Priority is not provided") + } + if mergeInto.Priority == nil { + mergeInto.Priority = &commonpb.Priority{} + } + mergeInto.Priority.FairnessWeight = mergeFrom.Priority.FairnessWeight + } + if mergeInto.RetryPolicy == nil { mergeInto.RetryPolicy = &commonpb.RetryPolicy{} } + if _, ok := updateFields["retryPolicy"]; ok { + mergeInto.RetryPolicy = mergeFrom.RetryPolicy + } + if _, ok := updateFields["retryPolicy.initialInterval"]; ok { if mergeFrom.RetryPolicy == nil { return serviceerror.NewInvalidArgument("RetryPolicy is not provided") @@ -295,6 +335,7 @@ func updateActivityOptions( activityInfo.ScheduleToStartTimeout = activityOptions.ScheduleToStartTimeout activityInfo.StartToCloseTimeout = activityOptions.StartToCloseTimeout activityInfo.HeartbeatTimeout = activityOptions.HeartbeatTimeout + activityInfo.Priority = activityOptions.Priority activityInfo.RetryMaximumInterval = activityOptions.RetryPolicy.MaximumInterval activityInfo.RetryBackoffCoefficient = activityOptions.RetryPolicy.BackoffCoefficient activityInfo.RetryInitialInterval = activityOptions.RetryPolicy.InitialInterval @@ -375,6 +416,7 @@ func restoreOriginalOptions( ScheduleToStartTimeout: originalOptions.ScheduleToStartTimeout, StartToCloseTimeout: originalOptions.StartToCloseTimeout, HeartbeatTimeout: originalOptions.HeartbeatTimeout, + Priority: originalOptions.Priority, RetryPolicy: originalOptions.RetryPolicy, } diff --git a/service/history/api/updateactivityoptions/api_test.go b/service/history/api/updateactivityoptions/api_test.go index 7f1aebd4059..d0b7ceb64c3 100644 --- a/service/history/api/updateactivityoptions/api_test.go +++ b/service/history/api/updateactivityoptions/api_test.go @@ -37,13 +37,17 @@ import ( ) func TestApplyActivityOptionsAcceptance(t *testing.T) { - - options := &activitypb.ActivityOptions{ + updateOptions := &activitypb.ActivityOptions{ TaskQueue: &taskqueuepb.TaskQueue{Name: "task_queue_name"}, ScheduleToCloseTimeout: durationpb.New(time.Second), StartToCloseTimeout: durationpb.New(time.Second), ScheduleToStartTimeout: durationpb.New(time.Second), HeartbeatTimeout: durationpb.New(time.Second), + Priority: &commonpb.Priority{ + PriorityKey: 42, + FairnessKey: "test_key", + FairnessWeight: 5.0, + }, RetryPolicy: &commonpb.RetryPolicy{ MaximumInterval: durationpb.New(time.Second), MaximumAttempts: 5, @@ -60,10 +64,10 @@ func TestApplyActivityOptionsAcceptance(t *testing.T) { mask *fieldmaskpb.FieldMask }{ { - name: "full mix - CamelCase", - mergeFrom: options, + name: "Top-level fields with CamelCase", + mergeFrom: updateOptions, mergeInto: &activitypb.ActivityOptions{}, - expected: options, + expected: updateOptions, mask: &fieldmaskpb.FieldMask{ Paths: []string{ "TaskQueue.Name", @@ -71,18 +75,16 @@ func TestApplyActivityOptionsAcceptance(t *testing.T) { "ScheduleToStartTimeout", "StartToCloseTimeout", "HeartbeatTimeout", - "RetryPolicy.BackoffCoefficient", - "RetryPolicy.InitialInterval", - "RetryPolicy.MaximumInterval", - "RetryPolicy.MaximumAttempts", + "Priority", + "RetryPolicy", }, }, }, { - name: "full mix - snake_case", - mergeFrom: options, + name: "Top-level fields with snake_case", + mergeFrom: updateOptions, mergeInto: &activitypb.ActivityOptions{}, - expected: options, + expected: updateOptions, mask: &fieldmaskpb.FieldMask{ Paths: []string{ "task_queue.name", @@ -90,38 +92,54 @@ func TestApplyActivityOptionsAcceptance(t *testing.T) { "schedule_to_start_timeout", "start_to_close_timeout", "heartbeat_timeout", - "retry_policy.backoff_coefficient", - "retry_policy.initial_interval", - "retry_policy.maximum_interval", - "retry_policy.maximum_attempts", + "priority", + "retry_policy", }, }, }, { - name: "partial", + name: "Sub-fields", mergeFrom: &activitypb.ActivityOptions{ - TaskQueue: &taskqueuepb.TaskQueue{Name: "task_queue_name"}, - ScheduleToCloseTimeout: durationpb.New(time.Second), - ScheduleToStartTimeout: durationpb.New(time.Second), + Priority: &commonpb.Priority{ + PriorityKey: 99, + FairnessKey: "newKey", + FairnessWeight: 7.5, + }, RetryPolicy: &commonpb.RetryPolicy{ - MaximumInterval: durationpb.New(time.Second), - MaximumAttempts: 5, + MaximumInterval: durationpb.New(time.Second), + MaximumAttempts: 5, + BackoffCoefficient: 1.0, + InitialInterval: durationpb.New(time.Second), }, }, mergeInto: &activitypb.ActivityOptions{ - StartToCloseTimeout: durationpb.New(time.Second), - HeartbeatTimeout: durationpb.New(time.Second), + Priority: &commonpb.Priority{ + PriorityKey: 10, + FairnessKey: "oldKey", + FairnessWeight: 1.0, + }, + RetryPolicy: &commonpb.RetryPolicy{}, + }, + expected: &activitypb.ActivityOptions{ + Priority: &commonpb.Priority{ + PriorityKey: 99, + FairnessKey: "newKey", + FairnessWeight: 7.5, + }, RetryPolicy: &commonpb.RetryPolicy{ + MaximumInterval: durationpb.New(time.Second), + MaximumAttempts: 5, BackoffCoefficient: 1.0, InitialInterval: durationpb.New(time.Second), }, }, - expected: options, mask: &fieldmaskpb.FieldMask{ Paths: []string{ - "task_queue.name", - "schedule_to_close_timeout", - "schedule_to_start_timeout", + "priority.priority_key", + "priority.fairness_key", + "priority.fairness_weight", + "retry_policy.backoff_coefficient", + "retry_policy.initial_interval", "retry_policy.maximum_interval", "retry_policy.maximum_attempts", }, @@ -145,7 +163,7 @@ func TestApplyActivityOptionsAcceptance(t *testing.T) { assert.Equal(t, tc.mergeInto.ScheduleToStartTimeout, tc.expected.ScheduleToStartTimeout, "ScheduleToStartTimeout") assert.Equal(t, tc.mergeInto.StartToCloseTimeout, tc.expected.StartToCloseTimeout, "StartToCloseTimeout") assert.Equal(t, tc.mergeInto.HeartbeatTimeout, tc.expected.HeartbeatTimeout, "HeartbeatTimeout") - + assert.Equal(t, tc.mergeInto.Priority, tc.expected.Priority, "Priority") } } @@ -153,23 +171,35 @@ func TestApplyActivityOptionsErrors(t *testing.T) { var err error err = mergeActivityOptions(&activitypb.ActivityOptions{}, &activitypb.ActivityOptions{}, util.ParseFieldMask(&fieldmaskpb.FieldMask{Paths: []string{"retry_policy.maximum_interval"}})) - assert.Error(t, err) + require.ErrorContains(t, err, "RetryPolicy is not provided") err = mergeActivityOptions(&activitypb.ActivityOptions{}, &activitypb.ActivityOptions{}, util.ParseFieldMask(&fieldmaskpb.FieldMask{Paths: []string{"retry_policy.maximum_attempts"}})) - assert.Error(t, err) + require.ErrorContains(t, err, "RetryPolicy is not provided") err = mergeActivityOptions(&activitypb.ActivityOptions{}, &activitypb.ActivityOptions{}, util.ParseFieldMask(&fieldmaskpb.FieldMask{Paths: []string{"retry_policy.backoff_coefficient"}})) - assert.Error(t, err) + require.ErrorContains(t, err, "RetryPolicy is not provided") err = mergeActivityOptions(&activitypb.ActivityOptions{}, &activitypb.ActivityOptions{}, util.ParseFieldMask(&fieldmaskpb.FieldMask{Paths: []string{"retry_policy.initial_interval"}})) - assert.Error(t, err) + require.ErrorContains(t, err, "RetryPolicy is not provided") err = mergeActivityOptions(&activitypb.ActivityOptions{}, &activitypb.ActivityOptions{}, util.ParseFieldMask(&fieldmaskpb.FieldMask{Paths: []string{"taskQueue.name"}})) - assert.Error(t, err) + require.ErrorContains(t, err, "TaskQueue is not provided") + + err = mergeActivityOptions(&activitypb.ActivityOptions{}, &activitypb.ActivityOptions{}, + util.ParseFieldMask(&fieldmaskpb.FieldMask{Paths: []string{"priority.priority_key"}})) + require.ErrorContains(t, err, "Priority is not provided") + + err = mergeActivityOptions(&activitypb.ActivityOptions{}, &activitypb.ActivityOptions{}, + util.ParseFieldMask(&fieldmaskpb.FieldMask{Paths: []string{"priority.fairness_key"}})) + require.ErrorContains(t, err, "Priority is not provided") + + err = mergeActivityOptions(&activitypb.ActivityOptions{}, &activitypb.ActivityOptions{}, + util.ParseFieldMask(&fieldmaskpb.FieldMask{Paths: []string{"priority.fairness_weight"}})) + require.ErrorContains(t, err, "Priority is not provided") } @@ -180,6 +210,11 @@ func TestApplyActivityOptionsReset(t *testing.T) { ScheduleToStartTimeout: durationpb.New(time.Second), StartToCloseTimeout: durationpb.New(time.Second), HeartbeatTimeout: durationpb.New(time.Second), + Priority: &commonpb.Priority{ + PriorityKey: 42, + FairnessKey: "test_key", + FairnessWeight: 5.0, + }, RetryPolicy: &commonpb.RetryPolicy{ MaximumInterval: durationpb.New(time.Second), MaximumAttempts: 5, @@ -194,6 +229,9 @@ func TestApplyActivityOptionsReset(t *testing.T) { "schedule_to_start_timeout", "start_to_close_timeout", "heartbeat_timeout", + "priority.priority_key", + "priority.fairness_key", + "priority.fairness_weight", "retry_policy.backoff_coefficient", "retry_policy.initial_interval", "retry_policy.maximum_interval", @@ -205,6 +243,9 @@ func TestApplyActivityOptionsReset(t *testing.T) { err := mergeActivityOptions(options, &activitypb.ActivityOptions{ + Priority: &commonpb.Priority{ + PriorityKey: 10, + }, RetryPolicy: &commonpb.RetryPolicy{ MaximumAttempts: 5, BackoffCoefficient: 1.0, @@ -218,6 +259,10 @@ func TestApplyActivityOptionsReset(t *testing.T) { assert.Nil(t, options.StartToCloseTimeout) assert.Nil(t, options.HeartbeatTimeout) + assert.Equal(t, int32(10), options.Priority.PriorityKey) + assert.Empty(t, options.Priority.FairnessKey) + assert.Zero(t, options.Priority.FairnessWeight) + assert.Nil(t, options.RetryPolicy.InitialInterval) assert.Nil(t, options.RetryPolicy.MaximumInterval) } diff --git a/service/history/api/updateworkflowoptions/api.go b/service/history/api/updateworkflowoptions/api.go index 0e225bf7e58..e3afe3c20d7 100644 --- a/service/history/api/updateworkflowoptions/api.go +++ b/service/history/api/updateworkflowoptions/api.go @@ -3,6 +3,7 @@ package updateworkflowoptions import ( "context" + commonpb "go.temporal.io/api/common/v1" "go.temporal.io/api/serviceerror" workflowpb "go.temporal.io/api/workflow/v1" "go.temporal.io/server/api/historyservice/v1" @@ -104,7 +105,7 @@ func MergeAndApply( if mergedOpts.GetVersioningOverride() == nil { unsetOverride = true } - _, err = ms.AddWorkflowExecutionOptionsUpdatedEvent(mergedOpts.GetVersioningOverride(), unsetOverride, "", nil, nil, identity) + _, err = ms.AddWorkflowExecutionOptionsUpdatedEvent(mergedOpts.GetVersioningOverride(), unsetOverride, "", nil, nil, identity, mergedOpts.GetPriority()) if err != nil { return nil, hasChanges, err } @@ -120,6 +121,11 @@ func getOptionsFromMutableState(ms historyi.MutableState) *workflowpb.WorkflowEx } opts.VersioningOverride = override } + if priority := ms.GetExecutionInfo().GetPriority(); priority != nil { + if cloned, ok := proto.Clone(priority).(*commonpb.Priority); ok { + opts.Priority = cloned + } + } return opts } @@ -150,5 +156,33 @@ func mergeWorkflowExecutionOptions( } mergeInto.VersioningOverride = mergeFrom.GetVersioningOverride() } + + // ==== Priority + + if _, ok := updateFields["priority"]; ok { + mergeInto.Priority = mergeFrom.GetPriority() + } + + if _, ok := updateFields["priority.priorityKey"]; ok { + if mergeInto.Priority == nil { + mergeInto.Priority = &commonpb.Priority{} + } + mergeInto.Priority.PriorityKey = mergeFrom.GetPriority().GetPriorityKey() + } + + if _, ok := updateFields["priority.fairnessKey"]; ok { + if mergeInto.Priority == nil { + mergeInto.Priority = &commonpb.Priority{} + } + mergeInto.Priority.FairnessKey = mergeFrom.Priority.GetFairnessKey() + } + + if _, ok := updateFields["priority.fairnessWeight"]; ok { + if mergeInto.Priority == nil { + mergeInto.Priority = &commonpb.Priority{} + } + mergeInto.Priority.FairnessWeight = mergeFrom.Priority.GetFairnessWeight() + } + return mergeInto, nil } diff --git a/service/history/api/updateworkflowoptions/api_test.go b/service/history/api/updateworkflowoptions/api_test.go index 951445c7b9e..51882ee556e 100644 --- a/service/history/api/updateworkflowoptions/api_test.go +++ b/service/history/api/updateworkflowoptions/api_test.go @@ -200,7 +200,7 @@ func (s *updateWorkflowOptionsSuite) TestInvoke_Success() { }, } s.currentMutableState.EXPECT().IsWorkflowExecutionRunning().Return(true) - s.currentMutableState.EXPECT().AddWorkflowExecutionOptionsUpdatedEvent(expectedOverrideOptions.VersioningOverride, false, "", nil, nil, "").Return(&historypb.HistoryEvent{}, nil) + s.currentMutableState.EXPECT().AddWorkflowExecutionOptionsUpdatedEvent(expectedOverrideOptions.VersioningOverride, false, "", nil, nil, "", expectedOverrideOptions.Priority).Return(&historypb.HistoryEvent{}, nil) s.currentContext.EXPECT().UpdateWorkflowExecutionAsActive(gomock.Any(), s.shardContext).Return(nil) updateReq := &historyservice.UpdateWorkflowExecutionOptionsRequest{ diff --git a/service/history/historybuilder/event_factory.go b/service/history/historybuilder/event_factory.go index 86365c2ee57..1e5d59c322c 100644 --- a/service/history/historybuilder/event_factory.go +++ b/service/history/historybuilder/event_factory.go @@ -385,6 +385,7 @@ func (b *EventFactory) CreateWorkflowExecutionOptionsUpdatedEvent( attachCompletionCallbacks []*commonpb.Callback, links []*commonpb.Link, identity string, + priority *commonpb.Priority, ) *historypb.HistoryEvent { event := b.createHistoryEvent(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED, b.timeSource.Now()) event.Attributes = &historypb.HistoryEvent_WorkflowExecutionOptionsUpdatedEventAttributes{ @@ -394,6 +395,7 @@ func (b *EventFactory) CreateWorkflowExecutionOptionsUpdatedEvent( AttachedRequestId: attachRequestID, AttachedCompletionCallbacks: attachCompletionCallbacks, Identity: identity, + Priority: priority, }, } event.Links = links diff --git a/service/history/historybuilder/history_builder.go b/service/history/historybuilder/history_builder.go index 5df04bc736f..002dadf0116 100644 --- a/service/history/historybuilder/history_builder.go +++ b/service/history/historybuilder/history_builder.go @@ -453,6 +453,7 @@ func (b *HistoryBuilder) AddWorkflowExecutionOptionsUpdatedEvent( attachCompletionCallbacks []*commonpb.Callback, links []*commonpb.Link, identity string, + priority *commonpb.Priority, ) *historypb.HistoryEvent { event := b.EventFactory.CreateWorkflowExecutionOptionsUpdatedEvent( worker_versioning.ConvertOverrideToV32(versioningOverride), @@ -461,6 +462,7 @@ func (b *HistoryBuilder) AddWorkflowExecutionOptionsUpdatedEvent( attachCompletionCallbacks, links, identity, + priority, ) event, _ = b.EventStore.add(event) return event diff --git a/service/history/historybuilder/history_builder_categorization_test.go b/service/history/historybuilder/history_builder_categorization_test.go index 45275714f33..3ae5619bf9f 100644 --- a/service/history/historybuilder/history_builder_categorization_test.go +++ b/service/history/historybuilder/history_builder_categorization_test.go @@ -208,7 +208,7 @@ func TestHistoryBuilder_FlushBufferToCurrentBatch(t *testing.T) { t.Errorf("expected 1 event in memBufferBatch got %d", len(hb.memBufferBatch)) } // add another event to memBufferBatch - hb.AddWorkflowExecutionOptionsUpdatedEvent(nil, false, "request-id-1", nil, nil, "") + hb.AddWorkflowExecutionOptionsUpdatedEvent(nil, false, "request-id-1", nil, nil, "", nil) if len(hb.memBufferBatch) != 2 { t.Errorf("expected 2 event in memBufferBatch got %d", len(hb.memBufferBatch)) } diff --git a/service/history/interfaces/mutable_state.go b/service/history/interfaces/mutable_state.go index 89d496d930e..aee318e279a 100644 --- a/service/history/interfaces/mutable_state.go +++ b/service/history/interfaces/mutable_state.go @@ -119,6 +119,7 @@ type ( attachCompletionCallbacks []*commonpb.Callback, links []*commonpb.Link, identity string, + priority *commonpb.Priority, ) (*historypb.HistoryEvent, error) AddWorkflowExecutionUpdateAcceptedEvent(protocolInstanceID string, acceptedRequestMessageId string, acceptedRequestSequencingEventId int64, acceptedRequest *updatepb.Request) (*historypb.HistoryEvent, error) AddWorkflowExecutionUpdateCompletedEvent(acceptedEventID int64, updResp *updatepb.Response) (*historypb.HistoryEvent, error) diff --git a/service/history/interfaces/mutable_state_mock.go b/service/history/interfaces/mutable_state_mock.go index ef97a9fd5e1..ef4f52ec631 100644 --- a/service/history/interfaces/mutable_state_mock.go +++ b/service/history/interfaces/mutable_state_mock.go @@ -639,18 +639,18 @@ func (mr *MockMutableStateMockRecorder) AddWorkflowExecutionCanceledEvent(arg0, } // AddWorkflowExecutionOptionsUpdatedEvent mocks base method. -func (m *MockMutableState) AddWorkflowExecutionOptionsUpdatedEvent(versioningOverride *workflow.VersioningOverride, unsetVersioningOverride bool, attachRequestID string, attachCompletionCallbacks []*common.Callback, links []*common.Link, identity string) (*history.HistoryEvent, error) { +func (m *MockMutableState) AddWorkflowExecutionOptionsUpdatedEvent(versioningOverride *workflow.VersioningOverride, unsetVersioningOverride bool, attachRequestID string, attachCompletionCallbacks []*common.Callback, links []*common.Link, identity string, priority *common.Priority) (*history.HistoryEvent, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "AddWorkflowExecutionOptionsUpdatedEvent", versioningOverride, unsetVersioningOverride, attachRequestID, attachCompletionCallbacks, links, identity) + ret := m.ctrl.Call(m, "AddWorkflowExecutionOptionsUpdatedEvent", versioningOverride, unsetVersioningOverride, attachRequestID, attachCompletionCallbacks, links, identity, priority) ret0, _ := ret[0].(*history.HistoryEvent) ret1, _ := ret[1].(error) return ret0, ret1 } // AddWorkflowExecutionOptionsUpdatedEvent indicates an expected call of AddWorkflowExecutionOptionsUpdatedEvent. -func (mr *MockMutableStateMockRecorder) AddWorkflowExecutionOptionsUpdatedEvent(versioningOverride, unsetVersioningOverride, attachRequestID, attachCompletionCallbacks, links, identity any) *gomock.Call { +func (mr *MockMutableStateMockRecorder) AddWorkflowExecutionOptionsUpdatedEvent(versioningOverride, unsetVersioningOverride, attachRequestID, attachCompletionCallbacks, links, identity, priority any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddWorkflowExecutionOptionsUpdatedEvent", reflect.TypeOf((*MockMutableState)(nil).AddWorkflowExecutionOptionsUpdatedEvent), versioningOverride, unsetVersioningOverride, attachRequestID, attachCompletionCallbacks, links, identity) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddWorkflowExecutionOptionsUpdatedEvent", reflect.TypeOf((*MockMutableState)(nil).AddWorkflowExecutionOptionsUpdatedEvent), versioningOverride, unsetVersioningOverride, attachRequestID, attachCompletionCallbacks, links, identity, priority) } // AddWorkflowExecutionPausedEvent mocks base method. diff --git a/service/history/ndc/events_reapplier_test.go b/service/history/ndc/events_reapplier_test.go index ecd8a3a23de..c2e013fb40c 100644 --- a/service/history/ndc/events_reapplier_test.go +++ b/service/history/ndc/events_reapplier_test.go @@ -111,7 +111,8 @@ func (s *nDCEventReapplicationSuite) TestReapplyEvents_AppliedEvent_WorkflowExec attr.GetAttachedRequestId(), attr.GetAttachedCompletionCallbacks(), event.Links, - event.GetWorkflowExecutionOptionsUpdatedEventAttributes().GetIdentity(), + attr.GetIdentity(), + attr.GetPriority(), ).Return(event, nil) msCurrent.EXPECT().HSM().Return(s.hsmNode).AnyTimes() msCurrent.EXPECT().IsWorkflowPendingOnWorkflowTaskBackoff().Return(true) diff --git a/service/history/ndc/workflow_resetter.go b/service/history/ndc/workflow_resetter.go index 4abf39ab31f..b52b0cdf6f4 100644 --- a/service/history/ndc/workflow_resetter.go +++ b/service/history/ndc/workflow_resetter.go @@ -954,6 +954,7 @@ func reapplyEvents( callbacks, event.Links, attr.GetIdentity(), + attr.GetPriority(), ); err != nil { return reappliedEvents, err } diff --git a/service/history/ndc/workflow_resetter_test.go b/service/history/ndc/workflow_resetter_test.go index 22b23e9d49e..4fec71a62e4 100644 --- a/service/history/ndc/workflow_resetter_test.go +++ b/service/history/ndc/workflow_resetter_test.go @@ -1193,6 +1193,7 @@ func (s *workflowResetterSuite) TestReapplyEvents() { attr.GetAttachedCompletionCallbacks(), event.Links, attr.GetIdentity(), + attr.GetPriority(), ).Return(&historypb.HistoryEvent{}, nil) case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED: attr := event.GetWorkflowExecutionSignaledEventAttributes() diff --git a/service/history/workflow/activity.go b/service/history/workflow/activity.go index e4034d95b98..fa600268256 100644 --- a/service/history/workflow/activity.go +++ b/service/history/workflow/activity.go @@ -224,6 +224,7 @@ func GetPendingActivityInfo( ScheduleToStartTimeout: ai.ScheduleToStartTimeout, StartToCloseTimeout: ai.StartToCloseTimeout, HeartbeatTimeout: ai.HeartbeatTimeout, + Priority: ai.Priority, RetryPolicy: &commonpb.RetryPolicy{ InitialInterval: ai.RetryInitialInterval, diff --git a/service/history/workflow/mutable_state_impl.go b/service/history/workflow/mutable_state_impl.go index d8265822c7c..432f53aaad2 100644 --- a/service/history/workflow/mutable_state_impl.go +++ b/service/history/workflow/mutable_state_impl.go @@ -5193,6 +5193,7 @@ func (ms *MutableStateImpl) AddWorkflowExecutionOptionsUpdatedEvent( attachCompletionCallbacks []*commonpb.Callback, links []*commonpb.Link, identity string, + priority *commonpb.Priority, ) (*historypb.HistoryEvent, error) { if err := ms.checkMutability(tag.WorkflowActionWorkflowOptionsUpdated); err != nil { return nil, err @@ -5204,6 +5205,7 @@ func (ms *MutableStateImpl) AddWorkflowExecutionOptionsUpdatedEvent( attachCompletionCallbacks, links, identity, + priority, ) prevEffectiveVersioningBehavior := ms.GetEffectiveVersioningBehavior() prevEffectiveDeployment := ms.GetEffectiveDeployment() @@ -5256,6 +5258,14 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionOptionsUpdatedEvent(event *his return err } + // Update priority. + if attributes.GetPriority() != nil { + if !proto.Equal(ms.executionInfo.Priority, attributes.GetPriority()) { + requestReschedulePendingWorkflowTask = true + } + ms.executionInfo.Priority = attributes.GetPriority() + } + // Finally, reschedule the pending workflow task if so requested. if requestReschedulePendingWorkflowTask { return ms.reschedulePendingWorkflowTask() diff --git a/service/history/workflow/mutable_state_impl_test.go b/service/history/workflow/mutable_state_impl_test.go index 71a8aff26c8..1c1a7d9b1d4 100644 --- a/service/history/workflow/mutable_state_impl_test.go +++ b/service/history/workflow/mutable_state_impl_test.go @@ -1126,7 +1126,7 @@ func (s *mutableStateSuite) TestOverride_UnpinnedBase_SetPinnedAndUnsetWithEmpty s.createMutableStateWithVersioningBehavior(baseBehavior, deployment1, tq) // set pinned override - event, err := s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(pinnedOptions2.GetVersioningOverride(), false, "", nil, nil, id) + event, err := s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(pinnedOptions2.GetVersioningOverride(), false, "", nil, nil, id, nil) s.NoError(err) s.verifyEffectiveDeployment(deployment2, overrideBehavior) s.verifyWorkflowOptionsUpdatedEventAttr( @@ -1141,7 +1141,7 @@ func (s *mutableStateSuite) TestOverride_UnpinnedBase_SetPinnedAndUnsetWithEmpty // unset pinned override with boolean id = uuid.NewString() - event, err = s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(nil, true, "", nil, nil, id) + event, err = s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(nil, true, "", nil, nil, id, nil) s.NoError(err) s.verifyEffectiveDeployment(deployment1, baseBehavior) s.verifyWorkflowOptionsUpdatedEventAttr( @@ -1163,7 +1163,7 @@ func (s *mutableStateSuite) TestOverride_PinnedBase_SetUnpinnedAndUnsetWithEmpty s.createMutableStateWithVersioningBehavior(baseBehavior, deployment1, tq) // set unpinned override - event, err := s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(unpinnedOptions.GetVersioningOverride(), false, "", nil, nil, id) + event, err := s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(unpinnedOptions.GetVersioningOverride(), false, "", nil, nil, id, nil) s.NoError(err) s.verifyEffectiveDeployment(deployment1, overrideBehavior) s.verifyWorkflowOptionsUpdatedEventAttr( @@ -1178,7 +1178,7 @@ func (s *mutableStateSuite) TestOverride_PinnedBase_SetUnpinnedAndUnsetWithEmpty // unset pinned override with empty id = uuid.NewString() - event, err = s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(nil, true, "", nil, nil, id) + event, err = s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(nil, true, "", nil, nil, id, nil) s.NoError(err) s.verifyEffectiveDeployment(deployment1, baseBehavior) s.verifyWorkflowOptionsUpdatedEventAttr( @@ -1199,7 +1199,7 @@ func (s *mutableStateSuite) TestOverride_RedirectFails() { id := uuid.NewString() s.createMutableStateWithVersioningBehavior(baseBehavior, deployment1, tq) - event, err := s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(pinnedOptions3.GetVersioningOverride(), false, "", nil, nil, id) + event, err := s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(pinnedOptions3.GetVersioningOverride(), false, "", nil, nil, id, nil) s.NoError(err) s.verifyEffectiveDeployment(deployment3, overrideBehavior) s.verifyWorkflowOptionsUpdatedEventAttr( @@ -1226,7 +1226,7 @@ func (s *mutableStateSuite) TestOverride_BaseDeploymentUpdatedOnCompletion() { id := uuid.NewString() s.createMutableStateWithVersioningBehavior(baseBehavior, deployment1, tq) - event, err := s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(pinnedOptions3.GetVersioningOverride(), false, "", nil, nil, id) + event, err := s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(pinnedOptions3.GetVersioningOverride(), false, "", nil, nil, id, nil) s.NoError(err) s.verifyEffectiveDeployment(deployment3, overrideBehavior) s.verifyWorkflowOptionsUpdatedEventAttr( @@ -1278,7 +1278,7 @@ func (s *mutableStateSuite) TestOverride_BaseDeploymentUpdatedOnCompletion() { // now we unset the override and check that the base deployment/behavior is in effect id = uuid.NewString() - event, err = s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(nil, true, "", nil, nil, id) + event, err = s.mutableState.AddWorkflowExecutionOptionsUpdatedEvent(nil, true, "", nil, nil, id, nil) s.NoError(err) s.verifyEffectiveDeployment(deployment2, baseBehavior) s.verifyWorkflowOptionsUpdatedEventAttr( diff --git a/service/matching/forwarder.go b/service/matching/forwarder.go index dcf9f846f72..c74edb363e7 100644 --- a/service/matching/forwarder.go +++ b/service/matching/forwarder.go @@ -130,8 +130,8 @@ func (fwdr *Forwarder) ForwardTask(ctx context.Context, task *internalTask) erro ScheduleToStartTimeout: expirationDuration, ForwardInfo: fwdr.getForwardInfo(task), VersionDirective: task.event.Data.GetVersionDirective(), - Priority: task.event.Data.GetPriority(), Stamp: task.event.Data.GetStamp(), + Priority: task.event.Data.GetPriority(), }, ) case enumspb.TASK_QUEUE_TYPE_ACTIVITY: @@ -147,8 +147,8 @@ func (fwdr *Forwarder) ForwardTask(ctx context.Context, task *internalTask) erro Clock: task.event.Data.GetClock(), ScheduleToStartTimeout: expirationDuration, ForwardInfo: fwdr.getForwardInfo(task), - Stamp: task.event.Data.GetStamp(), VersionDirective: task.event.Data.GetVersionDirective(), + Stamp: task.event.Data.GetStamp(), Priority: task.event.Data.GetPriority(), }, ) diff --git a/service/matching/pri_forwarder.go b/service/matching/pri_forwarder.go index 88c513d7d64..186a1c005e1 100644 --- a/service/matching/pri_forwarder.go +++ b/service/matching/pri_forwarder.go @@ -88,8 +88,8 @@ func (f *priForwarder) ForwardTask(ctx context.Context, task *internalTask) erro ScheduleToStartTimeout: expirationDuration, ForwardInfo: f.getForwardInfo(task), VersionDirective: task.event.Data.GetVersionDirective(), - Priority: task.event.Data.GetPriority(), Stamp: task.event.Data.GetStamp(), + Priority: task.event.Data.GetPriority(), }, ) case enumspb.TASK_QUEUE_TYPE_ACTIVITY: @@ -105,8 +105,8 @@ func (f *priForwarder) ForwardTask(ctx context.Context, task *internalTask) erro Clock: task.event.Data.GetClock(), ScheduleToStartTimeout: expirationDuration, ForwardInfo: f.getForwardInfo(task), - Stamp: task.event.Data.GetStamp(), VersionDirective: task.event.Data.GetVersionDirective(), + Stamp: task.event.Data.GetStamp(), Priority: task.event.Data.GetPriority(), }, ) diff --git a/tests/priority_fairness_test.go b/tests/priority_fairness_test.go index 2c360514bd0..e274914dfb3 100644 --- a/tests/priority_fairness_test.go +++ b/tests/priority_fairness_test.go @@ -11,19 +11,24 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" + activitypb "go.temporal.io/api/activity/v1" commandpb "go.temporal.io/api/command/v1" commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" taskqueuepb "go.temporal.io/api/taskqueue/v1" + workflowpb "go.temporal.io/api/workflow/v1" "go.temporal.io/api/workflowservice/v1" "go.temporal.io/server/api/adminservice/v1" taskqueuespb "go.temporal.io/server/api/taskqueue/v1" "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/payloads" + "go.temporal.io/server/common/serviceerror" "go.temporal.io/server/common/testing/taskpoller" "go.temporal.io/server/common/testing/testvars" "go.temporal.io/server/tests/testcore" "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/fieldmaskpb" ) type PrioritySuite struct { @@ -560,3 +565,194 @@ func (s *FairnessSuite) TestFairness_Migration_FromFair() { // fair->pri, pri->fair. fair metadata will be created first. s.testMigration(true, true) } + +func (s *FairnessSuite) TestFairness_UpdateWorkflowExecutionOptions_InvalidatesPendingTask() { + tv := testvars.New(s.T()) + + capture := s.GetTestCluster().Host().CaptureMetricsHandler().StartCapture() + defer s.GetTestCluster().Host().CaptureMetricsHandler().StopCapture(capture) + + ctx, cancel := context.WithTimeout(s.T().Context(), 10*time.Second) + defer cancel() + + originalPriority := &commonpb.Priority{FairnessKey: "KEY"} + updatedPriority := &commonpb.Priority{FairnessKey: "NEW_KEY"} + + // Queue up new workflow. + startResp, err := s.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{ + Namespace: s.Namespace().String(), + WorkflowId: tv.WorkflowID(), + WorkflowType: tv.WorkflowType(), + TaskQueue: tv.TaskQueue(), + Priority: originalPriority, + }) + s.NoError(err) + + // Wait for workflow task to be backlogged. + s.Eventually(func() bool { + resp, err := s.AdminClient().GetTaskQueueTasks(ctx, &adminservice.GetTaskQueueTasksRequest{ + Namespace: s.Namespace().String(), + TaskQueue: tv.TaskQueue().Name, + TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW, + BatchSize: 10, + MinPass: 1, + }) + return err == nil && len(resp.GetTasks()) == 1 + }, 10*time.Second, 100*time.Millisecond) + + // Update workflow options to set a new priority. + updateResp, err := s.FrontendClient().UpdateWorkflowExecutionOptions(ctx, &workflowservice.UpdateWorkflowExecutionOptionsRequest{ + Namespace: s.Namespace().String(), + WorkflowExecution: &commonpb.WorkflowExecution{ + WorkflowId: tv.WorkflowID(), + RunId: startResp.GetRunId(), + }, + WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{ + Priority: updatedPriority, + }, + UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"priority"}}, + }) + s.NoError(err) + s.NotNil(updateResp.GetWorkflowExecutionOptions()) + s.ProtoEqual(updatedPriority, updateResp.GetWorkflowExecutionOptions().GetPriority()) + + // Query workflow to verify workflow has the updated priority. + descResp, err := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{ + Namespace: s.Namespace().String(), + Execution: &commonpb.WorkflowExecution{ + WorkflowId: tv.WorkflowID(), + RunId: startResp.GetRunId(), + }, + }) + s.NoError(err) + s.NotNil(descResp.GetWorkflowExecutionInfo()) + s.NotNil(descResp.GetWorkflowExecutionInfo().GetPriority()) + s.ProtoEqual(updatedPriority, descResp.GetWorkflowExecutionInfo().GetPriority()) + + // Poll for workflow task and schedule an activity. + _, err = s.TaskPoller().PollAndHandleWorkflowTask( + tv, + func(task *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) { + s.NotNil(task) + s.ContainsHistoryEvents(` + 3 WorkflowExecutionOptionsUpdated { "Priority": { "FairnessKey": "NEW_KEY" } } + 4 WorkflowTaskStarted + `, task.History.Events) + + return &workflowservice.RespondWorkflowTaskCompletedRequest{ + Commands: []*commandpb.Command{ + { + CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK, + Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{ + ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{ + ActivityId: tv.ActivityID(), + ActivityType: tv.ActivityType(), + TaskQueue: tv.TaskQueue(), + Priority: originalPriority, + ScheduleToCloseTimeout: durationpb.New(time.Minute), + }, + }, + }, + }, + }, nil + }, + taskpoller.WithContext(ctx), + ) + s.NoError(err) + + // Verify that 2 workflow tasks were sent to matching; and 1 was marked obsolete. + addWorkflowTaskCount := 0 + obsoleteWorkflowTaskCount := 0 + snap := capture.Snapshot() + for _, rec := range snap[metrics.ClientRequests.Name()] { + for key, val := range rec.Tags { + if key == metrics.OperationTagName && val == metrics.MatchingClientAddWorkflowTaskScope { + addWorkflowTaskCount++ + } + } + } + s.Equal(2, addWorkflowTaskCount, "Expected 2 workflow tasks to be dispatched to matching") + for _, rec := range snap[metrics.ClientFailures.Name()] { + for key, val := range rec.Tags { + if key == metrics.ErrorTypeTagName && val == fmt.Sprintf("%T", serviceerror.ObsoleteMatchingTask{}) { + obsoleteWorkflowTaskCount++ + } + } + } + s.Equal(1, obsoleteWorkflowTaskCount, "Expected 1 worklow task to be obsolete") + + // Wait for activity task to be backlogged + s.Eventually(func() bool { + resp, err := s.AdminClient().GetTaskQueueTasks(ctx, &adminservice.GetTaskQueueTasksRequest{ + Namespace: s.Namespace().String(), + TaskQueue: tv.TaskQueue().Name, + TaskQueueType: enumspb.TASK_QUEUE_TYPE_ACTIVITY, + BatchSize: 10, + MinPass: 1, + }) + return err == nil && len(resp.GetTasks()) == 1 + }, 10*time.Second, 100*time.Millisecond) + + // Update activity options to set a new priority + _, err = s.FrontendClient().UpdateActivityOptions(ctx, &workflowservice.UpdateActivityOptionsRequest{ + Namespace: s.Namespace().String(), + Execution: &commonpb.WorkflowExecution{ + WorkflowId: tv.WorkflowID(), + RunId: startResp.GetRunId(), + }, + Activity: &workflowservice.UpdateActivityOptionsRequest_Id{Id: tv.ActivityID()}, + ActivityOptions: &activitypb.ActivityOptions{ + Priority: updatedPriority, + }, + UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"priority"}}, + }) + s.NoError(err) + + // Query workflow to verify activity has the updated priority. + descResp, err = s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{ + Namespace: s.Namespace().String(), + Execution: &commonpb.WorkflowExecution{ + WorkflowId: tv.WorkflowID(), + RunId: startResp.GetRunId(), + }, + }) + s.NoError(err) + s.NotNil(descResp.GetPendingActivities()) + s.Len(descResp.GetPendingActivities(), 1) + s.ProtoEqual(updatedPriority, descResp.GetPendingActivities()[0].GetPriority()) + s.ProtoEqual(updatedPriority, descResp.GetPendingActivities()[0].GetActivityOptions().GetPriority()) + + // Poll for activity task and verify it has the updated priority. + _, err = s.TaskPoller().PollAndHandleActivityTask( + tv, + func(task *workflowservice.PollActivityTaskQueueResponse) (*workflowservice.RespondActivityTaskCompletedRequest, error) { + s.NotNil(task) + s.Equal(tv.ActivityID(), task.ActivityId) + s.ProtoEqual(updatedPriority, task.Priority) + return &workflowservice.RespondActivityTaskCompletedRequest{}, nil + }, + taskpoller.WithContext(ctx), + ) + s.NoError(err) + + // Verify that 2 activity tasks were sent to matching; and 1 was marked obsolete + addActivityTaskCount := 0 + obsoleteActivityTaskCount := 0 + snap = capture.Snapshot() + for _, rec := range snap[metrics.ClientRequests.Name()] { + for key, val := range rec.Tags { + if key == metrics.OperationTagName && val == metrics.MatchingClientAddActivityTaskScope { + addActivityTaskCount++ + } + } + } + s.Equal(2, addActivityTaskCount, "Expected 2 activity tasks to be dispatched to matching") + for _, rec := range snap[metrics.ClientFailures.Name()] { + for key, val := range rec.Tags { + if key == metrics.ErrorTypeTagName && val == fmt.Sprintf("%T", serviceerror.ObsoleteMatchingTask{}) { + obsoleteActivityTaskCount++ + } + } + } + s.Equal(1+obsoleteWorkflowTaskCount, obsoleteActivityTaskCount, "Expected 1 activity task to be obsolete") +}