Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Fairness weight override API (#8379)

Browse Source 
## What changed?

Adds the ability to set/unset fairness weight overrides by fairness key
via `UpdateTaskQueueConfig` API.

From the API:

> // Fairness weight for a task can come from multiple sources for
    // flexibility. From highest to lowest precedence:
// 1. Weights for a small set of keys can be overridden in task queue
    //    configuration with an API.
    // 2. It can be attached to the workflow/activity in this field.
    // 3. The default weight of 1.0 will be used.
    //
    // Weight values are clamped to the range [0.001, 1000].

## Why?

To allow users to fine-tune their fairness key behavior via the API. 

## How did you test it?
- [x] built
- [x] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [x] added new functional test(s)

## Potential risks
N/A
This commit is contained in:
Stephan Behnke
2025-10-16 17:37:31 -07:00
committed by GitHub
parent 9afa18f821
commit 2f4e55081b
25 changed files with 614 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
common/dynamicconfig/constants.go
View File

@@ -1344,6 +1344,11 @@ second per poller by one physical queue manager`,
2000,
"Cache size for fairness key rate limits.",
)
MatchingMaxFairnessKeyWeightOverrides = NewTaskQueueIntSetting(
"matching.maxFairnessKeyWeightOverrides",
1000,
"Maximum number of fairness key weight overrides that can be configured for a task queue at a time.",
)
// keys for history

24
common/priorities/priority_util.go
View File

@@ -12,9 +12,9 @@ const (
)
var (
ErrInvalidPriority = serviceerror.NewInvalidArgument("PriorityKey can't be negative")
ErrFairnessKeyLength = serviceerror.NewInvalidArgument("FairnessKey length exceeds limit")
ErrInvalidFairnessWeight = serviceerror.NewInvalidArgument("FairnessWeight can't be negative")
ErrInvalidPriority = serviceerror.NewInvalidArgument("priority key can't be negative")
ErrFairnessKeyLength = serviceerror.NewInvalidArgument("fairness key length exceeds limit")
ErrInvalidFairnessWeight = serviceerror.NewInvalidArgument("must be greater than zero")
)
func Merge(
@@ -34,13 +34,27 @@ func Merge(
}
}
func ValidateFairnessKey(key string) error {
if len(key) > fairnessKeyMaxLength {
return ErrFairnessKeyLength
}
return nil
}
func ValidateFairnessWeight(weight float32) error {
if weight <= 0 {
return ErrInvalidFairnessWeight
}
return nil
}
func Validate(p *commonpb.Priority) error {
if p == nil {
return nil
} else if p.PriorityKey < 0 {
return ErrInvalidPriority
} else if len(p.FairnessKey) > fairnessKeyMaxLength {
return ErrFairnessKeyLength
} else if err := ValidateFairnessKey(p.FairnessKey); err != nil {
return err
} else if p.FairnessWeight < 0 {
return ErrInvalidFairnessWeight
}

2
go.mod
View File

@@ -59,7 +59,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.54.1-0.20251016004347-cf9c3e8c6ed7
go.temporal.io/api v1.55.0
go.temporal.io/sdk v1.35.0
go.uber.org/fx v1.24.0
go.uber.org/mock v0.6.0

4
go.sum
View File

@@ -393,8 +393,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.54.1-0.20251016004347-cf9c3e8c6ed7 h1:HnyFCXUWchiduyC1qav0Mt/w0q0Sva6t/JSa0Zw8jvQ=
go.temporal.io/api v1.54.1-0.20251016004347-cf9c3e8c6ed7/go.mod h1:iaxoP/9OXMJcQkETTECfwYq4cw/bj4nwov8b3ZLVnXM=
go.temporal.io/api v1.55.0 h1:bnMyQuVTwZKjzyz+gtlzOC6O0ZRJ9UOzxtHzq0XJAOk=
go.temporal.io/api v1.55.0/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=

4
service/frontend/service.go
View File

@@ -78,6 +78,8 @@ type Config struct {
ShutdownDrainDuration dynamicconfig.DurationPropertyFn
ShutdownFailHealthCheckDuration dynamicconfig.DurationPropertyFn
MaxFairnessWeightOverrideConfigLimit dynamicconfig.IntPropertyFnWithTaskQueueFilter
MaxBadBinaries dynamicconfig.IntPropertyFnWithNamespaceFilter
// security protection settings
@@ -318,6 +320,8 @@ func NewConfig(
DeleteNamespaceConcurrentDeleteExecutionsActivities: dynamicconfig.DeleteNamespaceConcurrentDeleteExecutionsActivities.Get(dc),
DeleteNamespaceNamespaceDeleteDelay: dynamicconfig.DeleteNamespaceNamespaceDeleteDelay.Get(dc),
MaxFairnessWeightOverrideConfigLimit: dynamicconfig.MatchingMaxFairnessKeyWeightOverrides.Get(dc),
EnableSchedules: dynamicconfig.FrontendEnableSchedules.Get(dc),
// [cleanup-wv-pre-release]

51
service/frontend/validators.go
View File

@@ -5,6 +5,11 @@ import (
commonpb "go.temporal.io/api/common/v1"
"go.temporal.io/api/serviceerror"
"go.temporal.io/api/workflowservice/v1"
"go.temporal.io/server/common/priorities"
)
var (
errFairnessKeyEmpty = serviceerror.NewInvalidArgument("fairness weight override key must not be empty")
)
func validateExecution(w *commonpb.WorkflowExecution) error {
@@ -39,3 +44,49 @@ func validateStringField(fieldName string, value string, maxLen int, required bo
}
return nil
}
func validateFairnessWeightUpdate(
set map[string]float32,
unset []string,
maxConfigLimit int,
) error {
total := len(set) + len(unset)
if total > maxConfigLimit {
return serviceerror.NewInvalidArgumentf(
"too many fairness weight overrides in request: got %d, maximum %d",
total, maxConfigLimit,
)
}
for k, w := range set {
if k == "" {
return errFairnessKeyEmpty
}
if err := priorities.ValidateFairnessKey(k); err != nil {
return err
}
if err := priorities.ValidateFairnessWeight(w); err != nil {
return serviceerror.NewInvalidArgumentf(
"invalid fairness weight weight for key %q: %v", k, err,
)
}
}
for _, k := range unset {
if k == "" {
return errFairnessKeyEmpty
}
if err := priorities.ValidateFairnessKey(k); err != nil {
return err
}
}
for _, u := range unset {
if _, ok := set[u]; ok {
return serviceerror.NewInvalidArgumentf(
"fairness weight override key %q present in both set and unset lists", u)
}
}
return nil
}

77
service/frontend/validators_test.go Normal file
View File

@@ -0,0 +1,77 @@
package frontend
import (
"strings"
"testing"
"github.com/stretchr/testify/require"
)
func TestValidateFairnessWeightUpdate(t *testing.T) {
t.Run("set overrides", func(t *testing.T) {
set := map[string]float32{
"a": 1.0,
"b": 2.3,
}
unset := []string{}
err := validateFairnessWeightUpdate(set, unset, 10)
require.NoError(t, err)
})
t.Run("unset overrides", func(t *testing.T) {
set := map[string]float32{}
unset := []string{"z"}
err := validateFairnessWeightUpdate(set, unset, 10)
require.NoError(t, err)
})
t.Run("enforce max number of overrides", func(t *testing.T) {
set := map[string]float32{
"a": 1.0,
}
unset := []string{"z"}
err := validateFairnessWeightUpdate(set, unset, 10)
require.NoError(t, err)
err = validateFairnessWeightUpdate(set, unset, 2)
require.NoError(t, err)
err = validateFairnessWeightUpdate(set, unset, 1)
require.ErrorContains(t, err, "too many fairness weight overrides in request: got 2, maximum 1")
})
t.Run("reject too long key in `set`", func(t *testing.T) {
set := map[string]float32{
strings.Repeat("abcdefg", 10): 1.0,
}
unset := []string{}
err := validateFairnessWeightUpdate(set, unset, 10)
require.ErrorContains(t, err, "fairness key length exceeds limit")
})
t.Run("reject too long key in `unset`", func(t *testing.T) {
set := map[string]float32{"a": 1.0}
unset := []string{strings.Repeat("abcdefg", 10)}
err := validateFairnessWeightUpdate(set, unset, 10)
require.ErrorContains(t, err, "fairness key length exceeds limit")
})
t.Run("reject negative weight", func(t *testing.T) {
set := map[string]float32{
"a": -2.0,
}
unset := []string{}
err := validateFairnessWeightUpdate(set, unset, 10)
require.ErrorContains(t, err, "invalid fairness weight weight for key \"a\": must be greater than zero")
})
t.Run("reject overlap between `set` and `unset`", func(t *testing.T) {
set := map[string]float32{
"a": 1.0,
}
unset := []string{"a"}
err := validateFairnessWeightUpdate(set, unset, 10)
require.ErrorContains(t, err, "fairness weight override key \"a\" present in both set and unset lists")
})
}

16
service/frontend/workflow_handler.go
View File

@@ -6005,23 +6005,35 @@ func (wh *WorkflowHandler) UpdateTaskQueueConfig(
if err != nil {
return nil, err
}
// Validation: prohibit setting rate limit on workflow task queues
if request.TaskQueueType == enumspb.TASK_QUEUE_TYPE_WORKFLOW {
return nil, serviceerror.NewInvalidArgument("Setting rate limit on workflow task queues is not allowed.")
}
queueRateLimit := request.GetUpdateQueueRateLimit()
fairnessKeyRateLimitDefault := request.GetUpdateFairnessKeyRateLimitDefault()
// Validate rate limits
queueRateLimit := request.GetUpdateQueueRateLimit()
if err := validateRateLimit(queueRateLimit, "UpdateQueueRateLimit"); err != nil {
return nil, err
}
fairnessKeyRateLimitDefault := request.GetUpdateFairnessKeyRateLimitDefault()
if err := validateRateLimit(fairnessKeyRateLimitDefault, "UpdateFairnessKeyRateLimitDefault"); err != nil {
return nil, err
}
// Validate identity field
if err := validateStringField("Identity", request.GetIdentity(), wh.config.MaxIDLengthLimit(), false); err != nil {
return nil, err
}
// Validate Fairness Weight Updates
setFairnessWeightOverrides := request.GetSetFairnessWeightOverrides()
unsetFairnessWeightOverrides := request.GetUnsetFairnessWeightOverrides()
limit := wh.config.MaxFairnessWeightOverrideConfigLimit(request.GetNamespace(), request.TaskQueue, request.TaskQueueType)
if err := validateFairnessWeightUpdate(setFairnessWeightOverrides, unsetFairnessWeightOverrides, limit); err != nil {
return nil, err
}
resp, err := wh.matchingClient.UpdateTaskQueueConfig(ctx, &matchingservice.UpdateTaskQueueConfigRequest{
NamespaceId: namespaceID.String(),
UpdateTaskqueueConfig: request,

1
service/matching/backlog_manager_test.go
View File

@@ -78,6 +78,7 @@ func (s *BacklogManagerTestSuite) SetupTest() {
s.ptqMgr = NewMockphysicalTaskQueueManager(s.controller)
s.ptqMgr.EXPECT().QueueKey().Return(queue).AnyTimes()
s.ptqMgr.EXPECT().ProcessSpooledTask(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
s.ptqMgr.EXPECT().GetFairnessWeightOverrides().AnyTimes().Return(fairnessWeightOverrides{ /* To avoid deadlock with gomock method */ })
var ctx context.Context
ctx, s.cancelCtx = context.WithCancel(context.Background())

6
service/matching/config.go
View File

@@ -82,6 +82,7 @@ type (
RateLimiterRefreshInterval time.Duration
FairnessKeyRateLimitCacheSize dynamicconfig.IntPropertyFnWithTaskQueueFilter
MaxFairnessKeyWeightOverrides dynamicconfig.IntPropertyFnWithTaskQueueFilter
// Time to hold a poll request before returning an empty response if there are no tasks
LongPollExpirationInterval dynamicconfig.DurationPropertyFnWithTaskQueueFilter
@@ -182,6 +183,7 @@ type (
// Rate limiting
RateLimiterRefreshInterval time.Duration
FairnessKeyRateLimitCacheSize func() int
MaxFairnessKeyWeightOverrides func() int
BreakdownMetricsByTaskQueue func() bool
BreakdownMetricsByPartition func() bool
@@ -296,6 +298,7 @@ func NewConfig(
PriorityLevels: dynamicconfig.MatchingPriorityLevels.Get(dc),
RateLimiterRefreshInterval: time.Minute,
FairnessKeyRateLimitCacheSize: dynamicconfig.MatchingFairnessKeyRateLimitCacheSize.Get(dc),
MaxFairnessKeyWeightOverrides: dynamicconfig.MatchingMaxFairnessKeyWeightOverrides.Get(dc),
MaxIDLengthLimit: dynamicconfig.MaxIDLengthLimit.Get(dc),
AdminNamespaceToPartitionDispatchRate: dynamicconfig.AdminMatchingNamespaceToPartitionDispatchRate.Get(dc),
@@ -442,6 +445,9 @@ func newTaskQueueConfig(tq *tqid.TaskQueue, config *Config, ns namespace.Name) *
FairnessKeyRateLimitCacheSize: func() int {
return config.FairnessKeyRateLimitCacheSize(ns.String(), taskQueueName, taskType)
},
MaxFairnessKeyWeightOverrides: func() int {
return config.MaxFairnessKeyWeightOverrides(ns.String(), taskQueueName, taskType)
},
PollerHistoryTTL: func() time.Duration {
return config.PollerHistoryTTL(ns.String())
},

4
service/matching/fair_task_writer.go
View File

@@ -113,8 +113,8 @@ func (w *fairTaskWriter) allocTaskIDs(reqs []*writeTaskRequest) error {
}
func (w *fairTaskWriter) pickPasses(tasks []*writeTaskRequest, bases []fairLevel) {
// TODO(fairness): get this from config
var overrides fairnessWeightOverrides
// Fetch latest fairness weight overrides from the partition's rate limit manager via pqMgr
overrides := w.backlogMgr.pqMgr.GetFairnessWeightOverrides()
for i, task := range tasks {
pri := task.taskInfo.Priority

32
service/matching/fairness_util.go
View File

@@ -1,6 +1,10 @@
package matching
import commonpb "go.temporal.io/api/common/v1"
import (
"maps"
commonpb "go.temporal.io/api/common/v1"
)
const (
// minWeight * strideFactor must be >= 1
@@ -24,3 +28,29 @@ func getEffectiveWeight(overrides fairnessWeightOverrides, pri *commonpb.Priorit
}
return weight
}
func mergeFairnessWeightOverrides(
existing fairnessWeightOverrides,
set fairnessWeightOverrides,
unset []string,
maxFairnessKeyWeightOverrides int,
) (fairnessWeightOverrides, error) {
if len(existing) == 0 {
// Validation already made sure that no keys of unset and set equal.
return set, nil
}
res := maps.Clone(existing)
for _, k := range unset {
delete(res, k)
}
maps.Copy(res, set)
if len(res) > maxFairnessKeyWeightOverrides {
return nil, errFairnessOverridesUpdateRejected
}
return res, nil
}

78
service/matching/fairness_util_test.go Normal file
View File

@@ -0,0 +1,78 @@
package matching
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestMergeFairnessWeightOverrides(t *testing.T) {
t.Run("apply upserts and deletes", func(t *testing.T) {
existing := fairnessWeightOverrides{"a": 1.0, "b": 2.0}
set := fairnessWeightOverrides{
"a": 3.0, // update
"c": 4.0, // insert
}
unset := []string{"b", "x"} // delete existing b and non-existent x (no-op)
out, err := mergeFairnessWeightOverrides(existing, set, unset, 10)
require.NoError(t, err)
require.Equal(t, fairnessWeightOverrides{"a": 3.0, "c": 4.0}, out)
})
t.Run("no update", func(t *testing.T) {
// nil set and unset
existing := fairnessWeightOverrides{"a": 1.2, "b": 3.4}
out, err := mergeFairnessWeightOverrides(existing, nil, nil, 10)
require.NoError(t, err)
require.Equal(t, existing, out)
// empty set and unset
existing = fairnessWeightOverrides{"a": 1.2, "b": 3.4}
out, err = mergeFairnessWeightOverrides(existing, fairnessWeightOverrides{}, []string{}, 10)
require.NoError(t, err)
require.Equal(t, existing, out)
// non-existent key in unset
existing = fairnessWeightOverrides{"a": 1.2, "b": 3.4}
out, err = mergeFairnessWeightOverrides(existing, fairnessWeightOverrides{}, []string{"does-not-exist"}, 2)
require.NoError(t, err)
require.Equal(t, existing, out)
// same key and value
existing = fairnessWeightOverrides{"a": 1.2, "b": 3.4}
out, err = mergeFairnessWeightOverrides(existing, fairnessWeightOverrides{"a": 1.2}, []string{}, 2)
require.NoError(t, err)
require.Equal(t, existing, out)
})
t.Run("return set when existing is empty", func(t *testing.T) {
set := fairnessWeightOverrides{"a": 1.2, "b": 3.4}
out, err := mergeFairnessWeightOverrides(nil, set, nil, 10)
require.NoError(t, err)
require.Equal(t, set, out)
out, err = mergeFairnessWeightOverrides(fairnessWeightOverrides{}, set, nil, 10)
require.NoError(t, err)
require.Equal(t, set, out)
})
t.Run("enforce capacity", func(t *testing.T) {
existing := fairnessWeightOverrides{"e": 1.0, "b": 2.0}
unset := []string{"b"} // remove one first -> size becomes 1
set := fairnessWeightOverrides{"a": 3.0, "c": 4.0} // adding two makes size 3 which exceeds capacity 2
out, err := mergeFairnessWeightOverrides(existing, set, unset, 2)
require.ErrorIs(t, err, errFairnessOverridesUpdateRejected)
require.Nil(t, out)
})
t.Run("check capacity after deletes", func(t *testing.T) {
existing := fairnessWeightOverrides{"a": 1.0, "b": 2.0}
unset := []string{"b"} // first, delete one
set := fairnessWeightOverrides{"c": 5.0} // then, add one
out, err := mergeFairnessWeightOverrides(existing, set, unset, 2)
require.NoError(t, err)
require.Equal(t, fairnessWeightOverrides{"a": 1.0, "c": 5.0}, out)
})
}

4
service/matching/matcher_data_test.go
View File

@@ -313,7 +313,7 @@ func (s *MatcherDataSuite) TestTaskForward() {
func (s *MatcherDataSuite) TestRateLimitedBacklog() {
s.md.rateLimitManager.SetEffectiveRPSAndSourceForTesting(10.0, enumspb.RATE_LIMIT_SOURCE_API)
s.md.rateLimitManager.UpdateSimpleRateLimitForTesting(300 * time.Millisecond)
s.md.rateLimitManager.UpdateSimpleRateLimitWithBurstForTesting(300 * time.Millisecond)
// register some backlog with old tasks
for i := range 100 {
@@ -352,7 +352,7 @@ func (s *MatcherDataSuite) TestRateLimitedBacklog() {
func (s *MatcherDataSuite) TestPerKeyRateLimit() {
s.md.rateLimitManager.SetFairnessKeyRateLimitDefaultForTesting(10.0, enumspb.RATE_LIMIT_SOURCE_API)
s.md.rateLimitManager.UpdatePerKeySimpleRateLimitForTesting(300 * time.Millisecond)
s.md.rateLimitManager.UpdatePerKeySimpleRateLimitWithBurstForTesting(300 * time.Millisecond)
// register some backlog with three keys
keys := []string{"key1", "key2", "key3"}
for i := range 300 {

23
service/matching/matching_engine.go
View File

@@ -3004,7 +3004,10 @@ func prepareTaskQueueUserData(
return data
}
func (e *matchingEngineImpl) UpdateTaskQueueConfig(ctx context.Context, request *matchingservice.UpdateTaskQueueConfigRequest) (*matchingservice.UpdateTaskQueueConfigResponse, error) {
func (e *matchingEngineImpl) UpdateTaskQueueConfig(
ctx context.Context,
request *matchingservice.UpdateTaskQueueConfigRequest,
) (*matchingservice.UpdateTaskQueueConfigResponse, error) {
taskQueueFamily, err := tqid.NewTaskQueueFamily(request.NamespaceId, request.UpdateTaskqueueConfig.GetTaskQueue())
if err != nil {
return nil, err
@@ -3040,18 +3043,36 @@ func (e *matchingEngineImpl) UpdateTaskQueueConfig(ctx context.Context, request
}
now := hlc.Next(existingClock, e.timeSource)
protoTs := hlc.ProtoTimestamp(now)
// Update relevant config fields
cfg := data.PerType[int32(taskQueueType)].Config
updateTaskQueueConfig := request.GetUpdateTaskqueueConfig()
updateIdentity := updateTaskQueueConfig.GetIdentity()
// Queue Rate Limit
if qrl := updateTaskQueueConfig.GetUpdateQueueRateLimit(); qrl != nil {
cfg.QueueRateLimit = buildRateLimitConfig(qrl, protoTs, updateIdentity)
}
// Fairness Queue Rate Limit
if fkrl := updateTaskQueueConfig.GetUpdateFairnessKeyRateLimitDefault(); fkrl != nil {
cfg.FairnessKeysRateLimitDefault = buildRateLimitConfig(fkrl, protoTs, updateIdentity)
}
// Fairness Weight Overrides
if len(updateTaskQueueConfig.GetSetFairnessWeightOverrides()) > 0 ||
len(updateTaskQueueConfig.GetUnsetFairnessWeightOverrides()) > 0 {
cfg.FairnessWeightOverrides, err = mergeFairnessWeightOverrides(
cfg.FairnessWeightOverrides,
updateTaskQueueConfig.GetSetFairnessWeightOverrides(),
updateTaskQueueConfig.GetUnsetFairnessWeightOverrides(),
tqm.GetConfig().MaxFairnessKeyWeightOverrides(),
)
if err != nil {
return nil, false, err
}
}
// Update the clock on TaskQueueUserData to enforce LWW on config updates
data.Clock = now
return data, true, nil

4
service/matching/physical_task_queue_manager.go
View File

@@ -689,6 +689,10 @@ func (c *physicalTaskQueueManagerImpl) MakePollerScalingDecision(
})
}
func (c *physicalTaskQueueManagerImpl) GetFairnessWeightOverrides() fairnessWeightOverrides {
return c.partitionMgr.GetRateLimitManager().GetFairnessWeightOverrides()
}
func (c *physicalTaskQueueManagerImpl) makePollerScalingDecisionImpl(
pollStartTime time.Time,
statsFn func() *taskqueuepb.TaskQueueStats,

2
service/matching/physical_task_queue_manager_interface.go
View File

@@ -54,5 +54,7 @@ type (
// MakePollerScalingDecision makes a decision on whether to scale pollers up or down based on the current state
// of the task queue and the task about to be returned.
MakePollerScalingDecision(pollStartTime time.Time) *taskqueuepb.PollerScalingDecision
// GetFairnessWeightOverrides returns current fairness weight overrides for this queue.
GetFairnessWeightOverrides() fairnessWeightOverrides
}
)

14
service/matching/physical_task_queue_manager_mock.go
View File

@@ -129,6 +129,20 @@ func (mr *MockphysicalTaskQueueManagerMockRecorder) GetAllPollerInfo() *gomock.C
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllPollerInfo", reflect.TypeOf((*MockphysicalTaskQueueManager)(nil).GetAllPollerInfo))
}
// GetFairnessWeightOverrides mocks base method.
func (m *MockphysicalTaskQueueManager) GetFairnessWeightOverrides() fairnessWeightOverrides {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetFairnessWeightOverrides")
ret0, _ := ret[0].(fairnessWeightOverrides)
return ret0
}
// GetFairnessWeightOverrides indicates an expected call of GetFairnessWeightOverrides.
func (mr *MockphysicalTaskQueueManagerMockRecorder) GetFairnessWeightOverrides() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetFairnessWeightOverrides", reflect.TypeOf((*MockphysicalTaskQueueManager)(nil).GetFairnessWeightOverrides))
}
// GetInternalTaskQueueStatus mocks base method.
func (m *MockphysicalTaskQueueManager) GetInternalTaskQueueStatus() []*taskqueue0.InternalTaskQueueStatus {
m.ctrl.T.Helper()

17
service/matching/ratelimit_manager.go
View File

@@ -163,10 +163,10 @@ func (r *rateLimitManager) computeAndApplyRateLimitLocked() {
// If the effective RPS has changed, we need to update the rate limiters.
if oldRPS != newRPS {
r.updateRatelimitLocked()
r.updateSimpleRateLimitLocked(defaultBurstDuration)
r.updateSimpleRateLimitWithBurstLocked(defaultBurstDuration)
}
// Internally, checks if the per-key rate limit has changed and updates it accordingly.
r.updatePerKeySimpleRateLimitLocked(defaultBurstDuration)
r.updatePerKeySimpleRateLimitWithBurstLocked(defaultBurstDuration)
}
// Lazy injection of poll metadata.
@@ -234,6 +234,8 @@ func (r *rateLimitManager) trySetRPSFromUserDataLocked() {
val := float64(fairnessKeyRateLimitDefault.GetRateLimit().GetRequestsPerSecond()) / float64(r.numReadPartitions)
r.fairnessKeyRateLimitDefault = &val
}
fairnessWeightOverrides := config.GetFairnessWeightOverrides()
r.perKeyOverrides = fairnessWeightOverrides
}
// updateRatelimitLocked checks and updates the overall queue rate limit if changed.
@@ -256,7 +258,7 @@ func (r *rateLimitManager) updateRatelimitLocked() {
}
// UpdateSimpleRateLimit updates the overall queue rate limits for the simpleRateLimiter implementation
func (r *rateLimitManager) updateSimpleRateLimitLocked(burstDuration time.Duration) {
func (r *rateLimitManager) updateSimpleRateLimitWithBurstLocked(burstDuration time.Duration) {
newRPS := r.effectiveRPS
r.wholeQueueLimit = makeSimpleLimiterParams(newRPS, burstDuration)
@@ -268,7 +270,7 @@ func (r *rateLimitManager) updateSimpleRateLimitLocked(burstDuration time.Durati
// UpdatePerKeySimpleRateLimit updates the per-key rate limit for the simpleRateLimit implementation
// UpdateTaskQueueConfig api is the single source for the per-key rate limit.
func (r *rateLimitManager) updatePerKeySimpleRateLimitLocked(burstDuration time.Duration) {
func (r *rateLimitManager) updatePerKeySimpleRateLimitWithBurstLocked(burstDuration time.Duration) {
if r.fairnessKeyRateLimitDefault == nil {
r.clearPerKeyRateLimitsLocked()
return
@@ -355,6 +357,13 @@ func (r *rateLimitManager) consumeTokens(now int64, task *internalTask, tokens i
}
}
// GetFairnessWeightOverrides returns the current fairness weight overrides.
func (r *rateLimitManager) GetFairnessWeightOverrides() fairnessWeightOverrides {
r.mu.Lock()
defer r.mu.Unlock()
return r.perKeyOverrides
}
func (r *rateLimitManager) Stop() {
for _, cancel := range r.cancels {
cancel()

10
service/matching/ratelimit_manager_test.go
View File

@@ -47,7 +47,7 @@ func (s *RateLimitManagerSuite) TestUpdatePerKeySimpleRateLimitLocked_WhenFairne
s.Equal(2, rateLimitManager.perKeyReady.Size())
s.True(rateLimitManager.perKeyLimit.limited())
// Update the per-key simple rate limit with fairnessKeyRateLimitDefault as nil
rateLimitManager.updatePerKeySimpleRateLimitLocked(time.Second)
rateLimitManager.updatePerKeySimpleRateLimitWithBurstLocked(time.Second)
// Verify that clearPerKeyRateLimitsLocked was called
// The cache should be replaced with a new empty cache
s.Equal(0, rateLimitManager.perKeyReady.Size(), "All per-key ready entries should be cleared")
@@ -57,16 +57,16 @@ func (s *RateLimitManagerSuite) TestUpdatePerKeySimpleRateLimitLocked_WhenFairne
// Additions to rateLimitManager for use by other unit tests:
func (r *rateLimitManager) UpdateSimpleRateLimitForTesting(burstDuration time.Duration) {
func (r *rateLimitManager) UpdateSimpleRateLimitWithBurstForTesting(burstDuration time.Duration) {
r.mu.Lock()
defer r.mu.Unlock()
r.updateSimpleRateLimitLocked(burstDuration)
r.updateSimpleRateLimitWithBurstLocked(burstDuration)
}
func (r *rateLimitManager) UpdatePerKeySimpleRateLimitForTesting(burstDuration time.Duration) {
func (r *rateLimitManager) UpdatePerKeySimpleRateLimitWithBurstForTesting(burstDuration time.Duration) {
r.mu.Lock()
defer r.mu.Unlock()
r.updatePerKeySimpleRateLimitLocked(burstDuration)
r.updatePerKeySimpleRateLimitWithBurstLocked(burstDuration)
}
func (r *rateLimitManager) SetAdminRateForTesting(rps float64) {

4
service/matching/task_queue_partition_manager.go
View File

@@ -576,6 +576,10 @@ func (pm *taskQueuePartitionManagerImpl) GetUserDataManager() userDataManager {
return pm.userDataManager
}
func (pm *taskQueuePartitionManagerImpl) GetConfig() *taskQueueConfig {
return pm.config
}
// GetAllPollerInfo returns all pollers that polled from this taskqueue in last few minutes
func (pm *taskQueuePartitionManagerImpl) GetAllPollerInfo() []*taskqueuepb.PollerInfo {
ret := pm.defaultQueue.GetAllPollerInfo()

1
service/matching/task_queue_partition_manager_interface.go
View File

@@ -63,5 +63,6 @@ type (
PutCache(key any, value any)
GetCache(key any) any
GetRateLimitManager() *rateLimitManager
GetConfig() *taskQueueConfig
}
)

14
service/matching/task_queue_partition_manager_mock.go
View File

@@ -148,6 +148,20 @@ func (mr *MocktaskQueuePartitionManagerMockRecorder) GetCache(key any) *gomock.C
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCache", reflect.TypeOf((*MocktaskQueuePartitionManager)(nil).GetCache), key)
}
// GetConfig mocks base method.
func (m *MocktaskQueuePartitionManager) GetConfig() *taskQueueConfig {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetConfig")
ret0, _ := ret[0].(*taskQueueConfig)
return ret0
}
// GetConfig indicates an expected call of GetConfig.
func (mr *MocktaskQueuePartitionManagerMockRecorder) GetConfig() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetConfig", reflect.TypeOf((*MocktaskQueuePartitionManager)(nil).GetConfig))
}
// GetRateLimitManager mocks base method.
func (m *MocktaskQueuePartitionManager) GetRateLimitManager() *rateLimitManager {
m.ctrl.T.Helper()

13
service/matching/user_data_manager.go
View File

@@ -98,11 +98,12 @@ type (
var _ userDataManager = (*userDataManagerImpl)(nil)
var (
errUserDataNoMutateNonRoot = serviceerror.NewInvalidArgument("can only mutate user data on root workflow task queue")
errRequestedVersionTooLarge = serviceerror.NewInvalidArgument("requested task queue user data for version greater than known version")
errTaskQueueClosed = serviceerror.NewUnavailable("task queue closed")
errUserDataUnmodified = errors.New("sentinel error for unchanged user data")
errUserDataVersionMismatch = errors.New("user data version mismatch")
errUserDataNoMutateNonRoot = serviceerror.NewInvalidArgument("can only mutate user data on root workflow task queue")
errRequestedVersionTooLarge = serviceerror.NewInvalidArgument("requested task queue user data for version greater than known version")
errTaskQueueClosed = serviceerror.NewUnavailable("task queue closed")
errFairnessOverridesUpdateRejected = serviceerror.NewInvalidArgument("fairness weight overrides update rejected: exceeding maximum key size")
errUserDataUnmodified = errors.New("sentinel error for unchanged user data")
errUserDataVersionMismatch = errors.New("user data version mismatch")
)
func newUserDataManager(
@@ -489,7 +490,7 @@ func (m *userDataManagerImpl) updateUserData(
return nil, false, serviceerror.NewFailedPreconditionf("user data version mismatch: requested: %d, current: %d", options.KnownVersion, preUpdateVersion)
}
updatedUserData, shouldReplicate, err := updateFn(preUpdateData)
if err == errUserDataUnmodified {
if err == errUserDataUnmodified || err == errFairnessOverridesUpdateRejected {
return userData, false, err
}
if err != nil {

367
tests/task_queue_test.go
View File

@@ -211,7 +211,6 @@ func (s *TaskQueueSuite) configureRateLimitAndLaunchWorkflows(
defer wg.Done()
mu.Lock()
*runTimes = append(*runTimes, time.Now())
time.Sleep(250 * time.Millisecond) //nolint
mu.Unlock()
return nil
}
@@ -233,6 +232,8 @@ func (s *TaskQueueSuite) configureRateLimitAndLaunchWorkflows(
activityWorker = worker.New(s.SdkClient(), activityTaskQueue, worker.Options{
// Setting rate limit at worker level (this will be ignored in favor of the limit set through the api)
TaskQueueActivitiesPerSecond: workerRPS,
// Setting rate limit to throttle the worker to 4 activities per second
WorkerActivitiesPerSecond: 4,
})
activityWorker.RegisterActivityWithOptions(activityFunc, activity.RegisterOptions{Name: activityName})
s.NoError(activityWorker.Start())
@@ -491,30 +492,20 @@ func (s *TaskQueueSuite) TestTaskQueueRateLimit_UpdateFromWorkerConfigAndAPI() {
firstGap, secondGap)
}
func (s *TaskQueueSuite) setupPerKeyRateLimitWorkflow(
ctx context.Context,
tv *testvars.TestVars,
keys []string,
tasksPerKey int,
wholeQueueRPS float64,
perKeyRPS float64,
) (map[string][]time.Time, []time.Time) {
func (s *TaskQueueSuite) TestWholeQueueLimit_TighterThanPerKeyDefault_IsEnforced() {
const (
wholeQueueRPS = 10.0 // tighter
perKeyRPS = 50.0 // looser than whole queue, should not bind
tasksPerKey = 30
buffer = 3 * time.Second // CI jitter
)
fairnessKeysWithWeight := map[string]float32{"A": 1.0, "B": 1.0, "C": 1.0}
tv := testvars.New(s.T())
total := len(keys) * tasksPerKey
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// Enable fairness
s.OverrideDynamicConfig(dynamicconfig.MatchingEnableFairness, true)
// Single partition for simplicity.
s.OverrideDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1)
s.OverrideDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 1)
// Fast refresh so limiter state picks up config quickly.
s.OverrideDynamicConfig(dynamicconfig.TaskQueueInfoByBuildIdTTL, 1*time.Millisecond)
// generate a unique base so IDs don't collide across shards/tests
base := uuid.NewString()
parsePattern := fmt.Sprintf("perkey-wf-%s-%%d", base) // for Sscanf later
// Set up the task queue config with whole-queue and per-key rate limits.
// configure task queue
_, err := s.FrontendClient().UpdateTaskQueueConfig(ctx, &workflowservice.UpdateTaskQueueConfigRequest{
Namespace: s.Namespace().String(),
Identity: tv.ClientIdentity(),
@@ -531,117 +522,13 @@ func (s *TaskQueueSuite) setupPerKeyRateLimitWorkflow(
})
s.NoError(err)
// Start workflows (each will schedule one activity tagged with a fairness key).
for i := range total {
wfID := fmt.Sprintf("perkey-wf-%s-%d", base, i)
_, err := s.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
Namespace: s.Namespace().String(),
WorkflowId: wfID,
WorkflowType: tv.WorkflowType(),
TaskQueue: tv.TaskQueue(),
})
s.NoError(err)
}
// Drain workflow tasks -> schedule activities with Priority.FairnessKey.
wfHandled := 0
for wfHandled < total {
if err := ctx.Err(); err != nil {
s.T().Fatalf("context deadline while draining workflow tasks: handled=%d/%d: %v", wfHandled, total, err)
}
_, err := s.TaskPoller().PollAndHandleWorkflowTask(
tv,
func(task *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
s.Equal(3, len(task.History.Events))
var idx int
_, scanErr := fmt.Sscanf(task.WorkflowExecution.WorkflowId, parsePattern, &idx)
s.NoError(scanErr)
key := keys[idx%len(keys)]
input, encErr := payloads.Encode(key)
s.NoError(encErr)
cmd := &commandpb.Command{
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK,
Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{
ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{
ActivityId: fmt.Sprintf("act-%d", idx),
ActivityType: tv.ActivityType(),
TaskQueue: tv.TaskQueue(),
ScheduleToCloseTimeout: durationpb.New(30 * time.Second),
Priority: &commonpb.Priority{FairnessKey: key},
Input: input,
},
},
}
return &workflowservice.RespondWorkflowTaskCompletedRequest{Commands: []*commandpb.Command{cmd}}, nil
},
taskpoller.WithContext(ctx),
)
if err == nil {
wfHandled++
}
}
s.Equal(total, wfHandled)
// Drain activity tasks, recording times.
perKeyTimes := make(map[string][]time.Time, len(keys))
for _, k := range keys {
perKeyTimes[k] = []time.Time{}
}
allTimes := make([]time.Time, 0, total)
actsHandled := 0
for actsHandled < total {
if err := ctx.Err(); err != nil {
s.T().Fatalf("context deadline while draining activity tasks: handled=%d/%d: %v", actsHandled, total, err)
}
_, err := s.TaskPoller().PollAndHandleActivityTask(
tv,
func(task *workflowservice.PollActivityTaskQueueResponse) (*workflowservice.RespondActivityTaskCompletedRequest, error) {
var key string
s.NoError(payloads.Decode(task.Input, &key))
now := time.Now()
perKeyTimes[key] = append(perKeyTimes[key], now)
allTimes = append(allTimes, now)
nothing, encErr := payloads.Encode()
s.NoError(encErr)
return &workflowservice.RespondActivityTaskCompletedRequest{Result: nothing}, nil
},
taskpoller.WithContext(ctx),
)
if err == nil {
actsHandled++
}
}
s.Equal(total, actsHandled)
// perKeyTimes : Used to verify that each key's activities are throttled correctly.
// allTimes : Used to verify the overall throughput of the task queue.
return perKeyTimes, allTimes
}
func (s *TaskQueueSuite) TestWholeQueueLimit_TighterThanPerKeyDefault_IsEnforced() {
const (
wholeQueueRPS = 10.0 // tighter
perKeyRPS = 50.0 // looser than whole queue, should not bind
tasksPerKey = 30
buffer = 3 * time.Second // CI jitter
)
keys := []string{"A", "B", "C"}
tv := testvars.New(s.T())
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
_, allTimes := s.setupPerKeyRateLimitWorkflow(ctx, tv, keys, tasksPerKey, wholeQueueRPS, perKeyRPS)
_, allTimes := s.runActivitiesWithPriorities(ctx, tv, fairnessKeysWithWeight, tasksPerKey)
// Measure overall throughput after initial burst, which should be limited by wholeQueueRPS.
start := allTimes[0]
end := allTimes[len(allTimes)-1]
expected := time.Duration(float64(tasksPerKey*len(keys))/wholeQueueRPS) * time.Second
expected := time.Duration(float64(tasksPerKey*len(fairnessKeysWithWeight))/wholeQueueRPS) * time.Second
actual := end.Sub(start)
s.T().Logf("Time taken for tasks across fairness keys to drain is %v vs expected %v", actual, expected)
@@ -659,16 +546,33 @@ func (s *TaskQueueSuite) TestPerKeyRateLimit_Default_IsEnforcedAcrossThreeKeys()
tasksPerKey = 30
buffer = 3 * time.Second // relax for CI jitter
)
keys := []string{"A", "B", "C"}
fairnessKeysWithWeight := map[string]float32{"A": 1.0, "B": 1.0, "C": 1.0}
tv := testvars.New(s.T())
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
perKeyTimes, _ := s.setupPerKeyRateLimitWorkflow(ctx, tv, keys, tasksPerKey, wholeQueueRPS, perKeyRPS)
// configure task queue
_, err := s.FrontendClient().UpdateTaskQueueConfig(ctx, &workflowservice.UpdateTaskQueueConfigRequest{
Namespace: s.Namespace().String(),
Identity: tv.ClientIdentity(),
TaskQueue: tv.TaskQueue().GetName(),
TaskQueueType: enumspb.TASK_QUEUE_TYPE_ACTIVITY,
UpdateQueueRateLimit: &workflowservice.UpdateTaskQueueConfigRequest_RateLimitUpdate{
RateLimit: &taskqueuepb.RateLimit{RequestsPerSecond: float32(wholeQueueRPS)},
Reason: "test: whole-queue limit",
},
UpdateFairnessKeyRateLimitDefault: &workflowservice.UpdateTaskQueueConfigRequest_RateLimitUpdate{
RateLimit: &taskqueuepb.RateLimit{RequestsPerSecond: float32(perKeyRPS)},
Reason: "test: per-key default",
},
})
s.NoError(err)
for _, key := range keys {
perKeyTimes, _ := s.runActivitiesWithPriorities(ctx, tv, fairnessKeysWithWeight, tasksPerKey)
for key := range fairnessKeysWithWeight {
times := perKeyTimes[key]
s.Len(times, tasksPerKey, "unexpected count for key %s", key)
@@ -688,10 +592,73 @@ func (s *TaskQueueSuite) TestPerKeyRateLimit_Default_IsEnforcedAcrossThreeKeys()
}
}
func (s *TaskQueueSuite) TestPerKeyRateLimit_WeightOverride_IsEnforcedAcrossThreeKeys() {
const (
perKeyRPS = 5.0 // base per-key limit
wholeQueueRPS = 1000.0 // keep high so only per-key gates
tasksPerKey = 30
buffer = 3 * time.Second
)
// Fairness key overrides take precedence over default.
// Override A and C to default and make B twice as heavy (ie ~2x effective RPS).
fairnessKeysWithWeight := map[string]float32{"A": 6666.0, "B": 0.0, "C": 6666.0} // defaults are opposite of override
fairnessWeightOverrides := map[string]float32{"A": 1.0, "B": 2.0, "C": 1.0}
tv := testvars.New(s.T())
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// configure task queue
_, err := s.FrontendClient().UpdateTaskQueueConfig(ctx, &workflowservice.UpdateTaskQueueConfigRequest{
Namespace: s.Namespace().String(),
Identity: tv.ClientIdentity(),
TaskQueue: tv.TaskQueue().GetName(),
TaskQueueType: enumspb.TASK_QUEUE_TYPE_ACTIVITY,
UpdateQueueRateLimit: &workflowservice.UpdateTaskQueueConfigRequest_RateLimitUpdate{
RateLimit: &taskqueuepb.RateLimit{RequestsPerSecond: float32(wholeQueueRPS)},
Reason: "test: whole-queue limit",
},
UpdateFairnessKeyRateLimitDefault: &workflowservice.UpdateTaskQueueConfigRequest_RateLimitUpdate{
RateLimit: &taskqueuepb.RateLimit{RequestsPerSecond: float32(perKeyRPS)},
Reason: "test: per-key default",
},
SetFairnessWeightOverrides: fairnessWeightOverrides,
})
s.NoError(err)
perKeyTimes, _ := s.runActivitiesWithPriorities(ctx, tv, fairnessKeysWithWeight, tasksPerKey)
for key, fairnessWeightOverride := range fairnessWeightOverrides {
times := perKeyTimes[key]
s.Len(times, tasksPerKey, "unexpected count for key %s", key)
start := times[0]
end := times[len(times)-1]
expected := time.Duration(float64(tasksPerKey)/(perKeyRPS*float64(fairnessWeightOverride))) * time.Second
actual := end.Sub(start)
s.T().Logf("Time taken for fairness key %s with weight %v to drain : %v vs expected %v",
key, fairnessWeightOverride, actual, expected)
s.GreaterOrEqual(actual, expected-buffer,
"per-key RPS violated for key %s with weight %v: actual %v < expected %v (-%v buffer)",
key, fairnessWeightOverride, actual, expected)
s.LessOrEqual(actual, expected+buffer,
"too slow for key %s with weight %v: actual %v > expected %v (+%v buffer)",
key, fairnessWeightOverride, actual, expected, buffer)
}
}
// TestUpdateAndDescribeTaskQueueConfig tests the update and describe task queue config functionality.
// It updates the task queue config via the frontend API and then describes the task queue to verify,
// that the updated configuration is reflected correctly.
func (s *TaskQueueSuite) TestUpdateAndDescribeTaskQueueConfig() {
// Enforce a smaller limit for max fairness key weight overrides for this test
s.OverrideDynamicConfig(dynamicconfig.MatchingMaxFairnessKeyWeightOverrides, 5)
// Send update.
tv := testvars.New(s.T())
taskQueueName := tv.TaskQueue().Name
namespace := s.Namespace().String()
@@ -699,6 +666,7 @@ func (s *TaskQueueSuite) TestUpdateAndDescribeTaskQueueConfig() {
updateRPS := float32(42)
updateReason := "frontend-update-test"
updateIdentity := "test-identity"
fairnessOverrides := map[string]float32{"k1": 1.0, "k2": 1.5, "k3": 2.0, "k4": 0.5, "k5": 3.0}
updateReq := &workflowservice.UpdateTaskQueueConfigRequest{
Namespace: namespace,
Identity: updateIdentity,
@@ -716,6 +684,7 @@ func (s *TaskQueueSuite) TestUpdateAndDescribeTaskQueueConfig() {
},
Reason: updateReason,
},
SetFairnessWeightOverrides: fairnessOverrides,
}
updateResp, err := s.FrontendClient().UpdateTaskQueueConfig(testcore.NewContext(), updateReq)
s.NoError(err)
@@ -727,7 +696,9 @@ func (s *TaskQueueSuite) TestUpdateAndDescribeTaskQueueConfig() {
s.Equal(updateRPS, updateResp.Config.FairnessKeysRateLimitDefault.RateLimit.RequestsPerSecond)
s.Equal(updateReason, updateResp.Config.FairnessKeysRateLimitDefault.Metadata.Reason)
s.Equal(updateIdentity, updateResp.Config.FairnessKeysRateLimitDefault.Metadata.UpdateIdentity)
s.Equal(fairnessOverrides, updateResp.Config.FairnessWeightOverrides)
// Request describe.
describeReq := &workflowservice.DescribeTaskQueueRequest{
Namespace: namespace,
TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueueName},
@@ -745,6 +716,24 @@ func (s *TaskQueueSuite) TestUpdateAndDescribeTaskQueueConfig() {
s.Equal(updateRPS, describeResp.Config.FairnessKeysRateLimitDefault.RateLimit.RequestsPerSecond)
s.Equal(updateReason, describeResp.Config.FairnessKeysRateLimitDefault.Metadata.Reason)
s.Equal(updateIdentity, updateResp.Config.FairnessKeysRateLimitDefault.Metadata.UpdateIdentity)
s.Equal(fairnessOverrides, describeResp.Config.FairnessWeightOverrides)
// Attempt to exceed the maximum allowed fairness weight overrides.
exceedReq := &workflowservice.UpdateTaskQueueConfigRequest{
Namespace: namespace,
Identity: updateIdentity,
TaskQueue: taskQueueName,
TaskQueueType: taskQueueType,
SetFairnessWeightOverrides: map[string]float32{"k6": 1.0}, // Exceeds the limit of 5
}
_, err = s.FrontendClient().UpdateTaskQueueConfig(testcore.NewContext(), exceedReq)
s.Error(err)
s.ErrorContains(err, "fairness weight overrides update rejected")
// Verify no change after rejected update.
describeResp, err = s.FrontendClient().DescribeTaskQueue(testcore.NewContext(), describeReq)
s.NoError(err)
s.Equal(fairnessOverrides, describeResp.Config.FairnessWeightOverrides)
}
func (s *TaskQueueSuite) TestUpdateUnsetAndDescribeTaskQueueConfig() {
@@ -831,3 +820,117 @@ func (s *TaskQueueSuite) TestUpdateUnsetAndDescribeTaskQueueConfig() {
s.Equal(unsetReasonFairness, describeResp.Config.FairnessKeysRateLimitDefault.Metadata.Reason)
s.Equal(updateIdentity, updateResp.Config.FairnessKeysRateLimitDefault.Metadata.UpdateIdentity)
}
func (s *TaskQueueSuite) runActivitiesWithPriorities(
ctx context.Context,
tv *testvars.TestVars,
fairnessKeysWithWeight map[string]float32,
activitiesPerKey int,
) (map[string][]time.Time, []time.Time) {
fairnessKeys := make([]string, 0, len(fairnessKeysWithWeight))
for k := range fairnessKeysWithWeight {
fairnessKeys = append(fairnessKeys, k)
}
total := len(fairnessKeys) * activitiesPerKey
// Enable fairness
s.OverrideDynamicConfig(dynamicconfig.MatchingEnableFairness, true)
// Single partition for simplicity.
s.OverrideDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1)
s.OverrideDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 1)
// generate a unique base so IDs don't collide across shards/tests
base := uuid.NewString()
parsePattern := fmt.Sprintf("perkey-wf-%s-%%d", base) // for Sscanf later
// Start workflows (each will schedule one activity tagged with a fairness key).
for i := range total {
wfID := fmt.Sprintf("perkey-wf-%s-%d", base, i)
_, err := s.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
Namespace: s.Namespace().String(),
WorkflowId: wfID,
WorkflowType: tv.WorkflowType(),
TaskQueue: tv.TaskQueue(),
})
s.NoError(err)
}
// Drain workflow tasks -> schedule activities with Priority.FairnessKey.
wfHandled := 0
for wfHandled < total {
if err := ctx.Err(); err != nil {
s.T().Fatalf("context deadline while draining workflow tasks: handled=%d/%d: %v", wfHandled, total, err)
}
_, err := s.TaskPoller().PollAndHandleWorkflowTask(
tv,
func(task *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
s.Len(task.History.Events, 3)
var idx int
_, scanErr := fmt.Sscanf(task.WorkflowExecution.WorkflowId, parsePattern, &idx)
s.NoError(scanErr)
key := fairnessKeys[idx%len(fairnessKeys)]
weight := fairnessKeysWithWeight[key]
input, encErr := payloads.Encode(key)
s.NoError(encErr)
cmd := &commandpb.Command{
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK,
Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{
ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{
ActivityId: fmt.Sprintf("act-%d", idx),
ActivityType: tv.ActivityType(),
TaskQueue: tv.TaskQueue(),
ScheduleToCloseTimeout: durationpb.New(30 * time.Second),
Priority: &commonpb.Priority{FairnessKey: key, FairnessWeight: weight},
Input: input,
},
},
}
return &workflowservice.RespondWorkflowTaskCompletedRequest{Commands: []*commandpb.Command{cmd}}, nil
},
taskpoller.WithContext(ctx),
)
if err == nil {
wfHandled++
}
}
s.Equal(total, wfHandled)
// Drain activity tasks, recording times.
perKeyTimes := make(map[string][]time.Time, len(fairnessKeys))
for _, k := range fairnessKeys {
perKeyTimes[k] = []time.Time{}
}
allTimes := make([]time.Time, 0, total)
actsHandled := 0
for actsHandled < total {
if err := ctx.Err(); err != nil {
s.T().Fatalf("context deadline while draining activity tasks: handled=%d/%d: %v", actsHandled, total, err)
}
_, err := s.TaskPoller().PollAndHandleActivityTask(
tv,
func(task *workflowservice.PollActivityTaskQueueResponse) (*workflowservice.RespondActivityTaskCompletedRequest, error) {
var key string
s.NoError(payloads.Decode(task.Input, &key))
now := time.Now()
perKeyTimes[key] = append(perKeyTimes[key], now)
allTimes = append(allTimes, now)
nothing, encErr := payloads.Encode()
s.NoError(encErr)
return &workflowservice.RespondActivityTaskCompletedRequest{Result: nothing}, nil
},
taskpoller.WithContext(ctx),
)
if err == nil {
actsHandled++
}
}
s.Equal(total, actsHandled)
// perKeyTimes : Used to verify that each key's activities are throttled correctly.
// allTimes : Used to verify the overall throughput of the task queue.
return perKeyTimes, allTimes
}