Add TaskQueueRateLimitFractionProvider interface to scale task queue rate limits (#10827)

## What changed?

Introduces a `TaskQueueRateLimitFractionProvider` interface in the
matching service that applies a fractional multiplier to the task queue
rate limit. The default implementation returns `1.0` (no change in
behavior). The interface can be overridden via `fx.Decorate` to supply a
custom fraction per namespace and task queue.

The provider is stored on `*Config` and bound as a closure
(`RateLimitFraction func() float64`) inside `newTaskQueueConfig`. The
`rateLimitManager` calls it directly in
`computeEffectiveRPSAndSourceLocked`. The `[0.0, 1.0]` range is enforced
at the provider layer via `NewTaskQueueRateLimitFractionProvider`.

Note: `adminNsRate` / `adminTqRate` are per-partition system caps and
are intentionally not scaled by the fraction.

## Why?

Provides an extension point for deployments that need to scale task
queue rate limits by a fractional value per namespace/task queue,
without changing rate limit logic.

## How did you test it?

- [x] built
- [ ] run locally and tested manually
- [x] covered by existing tests
- [x] added new unit tests for clamping behavior

## Potential risks

The fraction defaults to `1.0` so existing behavior is unchanged. A
misconfigured implementation returning `0.0` would effectively pause
task dispatch — operators overriding this interface should ensure
correctness of the returned fraction.

---------

Signed-off-by: Sandeep Balaji <sandeep.balaji@temporal.io>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: cbsandeep10

service/matching/config.go

@@ -96,6 +96,8 @@ type (
 		TaskQueueInfoByBuildIdTTL                dynamicconfig.DurationPropertyFnWithTaskQueueFilter
 		PriorityLevels                           dynamicconfig.IntPropertyFnWithTaskQueueFilter
 
+		RateLimitFractionProvider TaskQueueRateLimitFractionProvider
+
 		RateLimiterRefreshInterval    time.Duration
 		FairnessKeyRateLimitCacheSize dynamicconfig.IntPropertyFnWithTaskQueueFilter
 		MaxFairnessKeyWeightOverrides dynamicconfig.IntPropertyFnWithTaskQueueFilter
@@ -214,6 +216,7 @@ type (
 		MaxVersionsInTaskQueue    func() int
 
 		// Rate limiting
+		RateLimitFraction             func() float64
 		RateLimiterRefreshInterval    time.Duration
 		FairnessKeyRateLimitCacheSize func() int
 		MaxFairnessKeyWeightOverrides func() int
@@ -387,6 +390,8 @@ func NewConfig(
 		PartitionScaleManagerSettings: dynamicconfig.MatchingPartitionScaleManager.Get(dc),
 
 		LogAllReqErrors: dynamicconfig.LogAllReqErrors.Get(dc),
+
+		RateLimitFractionProvider: defaultTaskQueueRateLimitFractionProvider,
 	}
 }
 
@@ -535,6 +540,9 @@ func newTaskQueueConfig(tq *tqid.TaskQueue, config *Config, ns namespace.Name) *
 		TaskQueueInfoByBuildIdTTL: func() time.Duration {
 			return config.TaskQueueInfoByBuildIdTTL(ns.String(), taskQueueName, taskType)
 		},
+		RateLimitFraction: func() float64 {
+			return config.RateLimitFractionProvider.GetRateLimitFraction(ns, taskQueueName, taskType)
+		},
 		RateLimiterRefreshInterval: config.RateLimiterRefreshInterval,
 		FairnessKeyRateLimitCacheSize: func() int {
 			return config.FairnessKeyRateLimitCacheSize(ns.String(), taskQueueName, taskType)

service/matching/fx.go

@@ -51,6 +51,7 @@ var Module = fx.Options(
 	fx.Provide(ServerProvider),
 	fx.Provide(NewService),
 	fx.Provide(simplePartitionScalerFactoryProvider),
+	fx.Provide(taskQueueRateLimitFractionProviderProvider),
 	fx.Invoke(ServiceLifetimeHooks),
 )
 
@@ -61,8 +62,11 @@ func ServerProvider(grpcServerOptions []grpc.ServerOption) *grpc.Server {
 func ConfigProvider(
 	dc *dynamicconfig.Collection,
 	persistenceConfig config.Persistence,
+	rateLimitFractionProvider TaskQueueRateLimitFractionProvider,
 ) *Config {
-	return NewConfig(dc)
+	cfg := NewConfig(dc)
+	cfg.RateLimitFractionProvider = rateLimitFractionProvider
+	return cfg
 }
 
 func ServiceErrorInterceptorProvider(

service/matching/matching_engine_test.go

@@ -245,7 +245,8 @@ func newMatchingEngine(
 		taskManager:     taskMgr,
 		fairTaskManager: fairTaskMgr,
 		historyClient:   mockHistoryClient,
-		partitions:      make(map[tqid.PartitionKey]taskQueuePartitionManager),
+
+		partitions: make(map[tqid.PartitionKey]taskQueuePartitionManager),
 		gaugeMetrics: gaugeMetrics{
 			loadedTaskQueueFamilyCount:    make(map[taskQueueCounterKey]int),
 			loadedTaskQueueCount:          make(map[taskQueueCounterKey]int),

service/matching/rate_limit_fraction_provider.go

@@ -0,0 +1,44 @@
+package matching
+
+import (
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/server/common/namespace"
+)
+
+// TaskQueueRateLimitFractionProvider returns the fraction of the rate limit for a task queue.
+// The returned fraction is always clamped to [0.0, 1.0].
+type TaskQueueRateLimitFractionProvider interface {
+	GetRateLimitFraction(nsName namespace.Name, tqName string, tqType enumspb.TaskQueueType) float64
+}
+
+const (
+	minRateLimitFraction     = 0.0
+	maxRateLimitFraction     = 1.0
+	defaultRateLimitFraction = maxRateLimitFraction
+)
+
+// TaskQueueRateLimitFractionProviderFunc is a function that implements TaskQueueRateLimitFractionProvider.
+type TaskQueueRateLimitFractionProviderFunc func(nsName namespace.Name, tqName string, tqType enumspb.TaskQueueType) float64
+
+func (f TaskQueueRateLimitFractionProviderFunc) GetRateLimitFraction(nsName namespace.Name, tqName string, tqType enumspb.TaskQueueType) float64 {
+	return f(nsName, tqName, tqType)
+}
+
+// NewTaskQueueRateLimitFractionProvider wraps inner and enforces [0.0, 1.0] on every call.
+func NewTaskQueueRateLimitFractionProvider(inner TaskQueueRateLimitFractionProvider) TaskQueueRateLimitFractionProvider {
+	return TaskQueueRateLimitFractionProviderFunc(func(nsName namespace.Name, tqName string, tqType enumspb.TaskQueueType) float64 {
+		return max(min(inner.GetRateLimitFraction(nsName, tqName, tqType), maxRateLimitFraction), minRateLimitFraction)
+	})
+}
+
+type unitRateLimitFractionProvider struct{}
+
+func (p *unitRateLimitFractionProvider) GetRateLimitFraction(_ namespace.Name, _ string, _ enumspb.TaskQueueType) float64 {
+	return defaultRateLimitFraction
+}
+
+var defaultTaskQueueRateLimitFractionProvider = NewTaskQueueRateLimitFractionProvider(&unitRateLimitFractionProvider{})
+
+func taskQueueRateLimitFractionProviderProvider() TaskQueueRateLimitFractionProvider {
+	return defaultTaskQueueRateLimitFractionProvider
+}

service/matching/rate_limit_fraction_provider_test.go

@@ -0,0 +1,27 @@
+package matching
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/server/common/namespace"
+)
+
+func TestNewTaskQueueRateLimitFractionProvider_ClampedToMax(t *testing.T) {
+	inner := TaskQueueRateLimitFractionProviderFunc(func(_ namespace.Name, _ string, _ enumspb.TaskQueueType) float64 { return 2.0 })
+	p := NewTaskQueueRateLimitFractionProvider(inner)
+	require.InEpsilon(t, maxRateLimitFraction, p.GetRateLimitFraction("ns", "tq", enumspb.TASK_QUEUE_TYPE_WORKFLOW), 0)
+}
+
+func TestNewTaskQueueRateLimitFractionProvider_ClampedToMin(t *testing.T) {
+	inner := TaskQueueRateLimitFractionProviderFunc(func(_ namespace.Name, _ string, _ enumspb.TaskQueueType) float64 { return -0.5 })
+	p := NewTaskQueueRateLimitFractionProvider(inner)
+	require.InDelta(t, minRateLimitFraction, p.GetRateLimitFraction("ns", "tq", enumspb.TASK_QUEUE_TYPE_WORKFLOW), 0)
+}
+
+func TestNewTaskQueueRateLimitFractionProvider_PassthroughInRange(t *testing.T) {
+	inner := TaskQueueRateLimitFractionProviderFunc(func(_ namespace.Name, _ string, _ enumspb.TaskQueueType) float64 { return 0.5 })
+	p := NewTaskQueueRateLimitFractionProvider(inner)
+	require.InEpsilon(t, 0.5, p.GetRateLimitFraction("ns", "tq", enumspb.TASK_QUEUE_TYPE_WORKFLOW), 0)
+}

service/matching/ratelimit_manager.go

@@ -63,7 +63,8 @@ const (
 )
 
 // Create a new rate limit manager for the task queue partition.
-func newRateLimitManager(userDataManager userDataManager,
+func newRateLimitManager(
+	userDataManager userDataManager,
 	config *taskQueueConfig,
 	taskQueueType enumspb.TaskQueueType,
 ) *rateLimitManager {
@@ -143,12 +144,13 @@ func (r *rateLimitManager) computeEffectiveRPSAndSourceLocked() {
 		r.adminTqRate,
 	)
 	r.systemRPS = systemRPS
+	fraction := r.config.RateLimitFraction()
 	switch {
 	case r.apiConfigRPS != nil:
-		effectiveRPS = *r.apiConfigRPS / float64(r.numReadPartitions)
+		effectiveRPS = *r.apiConfigRPS * fraction / float64(r.numReadPartitions)
 		rateLimitSource = enumspb.RATE_LIMIT_SOURCE_API
 	case r.workerRPS != nil:
-		effectiveRPS = *r.workerRPS / float64(r.numReadPartitions)
+		effectiveRPS = *r.workerRPS * fraction / float64(r.numReadPartitions)
 		rateLimitSource = enumspb.RATE_LIMIT_SOURCE_WORKER
 	}
 
@@ -235,8 +237,10 @@ func (r *rateLimitManager) trySetRPSFromUserDataLocked() {
 	if fairnessKeyRateLimitDefault.GetRateLimit() == nil {
 		r.fairnessKeyRateLimitDefault = nil
 	} else {
-		// Maintain the fairnessKeyRateLimitDefault as per-partition rate.
-		val := float64(fairnessKeyRateLimitDefault.GetRateLimit().GetRequestsPerSecond()) / float64(r.numReadPartitions)
+		// Maintain the fairnessKeyRateLimitDefault as per-partition rate, scaled by the same
+		// fraction applied to the whole-queue effectiveRPS.
+		fraction := r.config.RateLimitFraction()
+		val := float64(fairnessKeyRateLimitDefault.GetRateLimit().GetRequestsPerSecond()) * fraction / float64(r.numReadPartitions)
 		r.fairnessKeyRateLimitDefault = &val
 	}
 	fairnessWeightOverrides := config.GetFairnessWeightOverrides()

service/matching/ratelimit_manager_test.go

@@ -7,7 +7,10 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	enumspb "go.temporal.io/api/enums/v1"
+	taskqueuepb "go.temporal.io/api/taskqueue/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common/dynamicconfig"
+	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/tqid"
 )
 
@@ -89,3 +92,137 @@ func (r *rateLimitManager) SetFairnessKeyRateLimitDefaultForTesting(rps float64,
 	defer r.mu.Unlock()
 	r.fairnessKeyRateLimitDefault = &rps
 }
+
+func (r *rateLimitManager) SetAPIConfigRPSForTesting(rps float64) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.apiConfigRPS = &rps
+	r.computeEffectiveRPSAndSourceLocked()
+}
+
+func (r *rateLimitManager) SetWorkerRPSForTesting(rps float64) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.workerRPS = &rps
+	r.computeEffectiveRPSAndSourceLocked()
+}
+
+// GetFairnessKeyRateLimitDefaultForTesting returns the stored per-partition fairness-key rate and
+// whether it is set. It is the value stored after fraction and partition scaling.
+func (r *rateLimitManager) GetFairnessKeyRateLimitDefaultForTesting() (float64, bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.fairnessKeyRateLimitDefault == nil {
+		return 0, false
+	}
+	return *r.fairnessKeyRateLimitDefault, true
+}
+
+func (r *rateLimitManager) GetNumReadPartitionsForTesting() int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.numReadPartitions
+}
+
+// newConfigWithFraction returns a Config whose RateLimitFractionProvider always returns the given fraction.
+func newConfigWithFraction(fraction float64) *Config {
+	cfg := NewConfig(dynamicconfig.NewNoopCollection())
+	cfg.RateLimitFractionProvider = TaskQueueRateLimitFractionProviderFunc(
+		func(_ namespace.Name, _ string, _ enumspb.TaskQueueType) float64 {
+			return fraction
+		},
+	)
+	return cfg
+}
+
+// TestFractionScaling_ApiConfigRPS verifies that apiConfigRPS is scaled by the fraction
+// before being divided by the number of read partitions.
+func (s *RateLimitManagerSuite) TestFractionScaling_ApiConfigRPS() {
+	cfg := newConfigWithFraction(0.5)
+	config := newTaskQueueConfig(
+		tqid.UnsafeTaskQueueFamily("test-ns", "test-tq").TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW),
+		cfg, "test-ns",
+	)
+	rlm := newRateLimitManager(&mockUserDataManager{}, config, enumspb.TASK_QUEUE_TYPE_WORKFLOW)
+	defer rlm.Stop()
+
+	rlm.SetAPIConfigRPSForTesting(100.0)
+
+	rps, source := rlm.GetEffectiveRPSAndSource()
+	// effectiveRPS = apiRPS * fraction / numPartitions; GetEffectiveRPSAndSource multiplies back by numPartitions → apiRPS * fraction
+	s.InEpsilon(50.0, rps, 1e-9)
+	s.Equal(enumspb.RATE_LIMIT_SOURCE_API, source)
+}
+
+// TestFractionScaling_WorkerRPS verifies that workerRPS is scaled by the fraction.
+func (s *RateLimitManagerSuite) TestFractionScaling_WorkerRPS() {
+	cfg := newConfigWithFraction(0.5)
+	config := newTaskQueueConfig(
+		tqid.UnsafeTaskQueueFamily("test-ns", "test-tq").TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW),
+		cfg, "test-ns",
+	)
+	rlm := newRateLimitManager(&mockUserDataManager{}, config, enumspb.TASK_QUEUE_TYPE_WORKFLOW)
+	defer rlm.Stop()
+
+	rlm.SetWorkerRPSForTesting(100.0)
+
+	rps, source := rlm.GetEffectiveRPSAndSource()
+	s.InEpsilon(50.0, rps, 1e-9)
+	s.Equal(enumspb.RATE_LIMIT_SOURCE_WORKER, source)
+}
+
+// TestFractionScaling_FairnessKeyRateLimitDefault verifies that the per-key fairness rate default
+// is also scaled by the fraction (Fix 3 coverage).
+func (s *RateLimitManagerSuite) TestFractionScaling_FairnessKeyRateLimitDefault() {
+	const fraction = 0.5
+	const fairnessKeyRPS = float32(100)
+	cfg := newConfigWithFraction(fraction)
+	config := newTaskQueueConfig(
+		tqid.UnsafeTaskQueueFamily("test-ns", "test-tq").TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW),
+		cfg, "test-ns",
+	)
+	udm := &mockUserDataManager{
+		data: &persistencespb.VersionedTaskQueueUserData{
+			Data: &persistencespb.TaskQueueUserData{
+				PerType: map[int32]*persistencespb.TaskQueueTypeUserData{
+					int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW): {
+						Config: &taskqueuepb.TaskQueueConfig{
+							FairnessKeysRateLimitDefault: &taskqueuepb.RateLimitConfig{
+								RateLimit: &taskqueuepb.RateLimit{RequestsPerSecond: fairnessKeyRPS},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	rlm := newRateLimitManager(udm, config, enumspb.TASK_QUEUE_TYPE_WORKFLOW)
+	defer rlm.Stop()
+
+	rlm.UserDataChanged()
+
+	got, ok := rlm.GetFairnessKeyRateLimitDefaultForTesting()
+	s.True(ok)
+	// val = fairnessKeyRPS * fraction / numPartitions
+	numPartitions := rlm.GetNumReadPartitionsForTesting()
+	want := float64(fairnessKeyRPS) * fraction / float64(numPartitions)
+	s.InEpsilon(want, got, 1e-9)
+}
+
+// TestFractionScaling_ZeroFraction verifies that a zero fraction (fully drained) reduces
+// the effective RPS to zero even when an API rate limit is set.
+func (s *RateLimitManagerSuite) TestFractionScaling_ZeroFraction() {
+	cfg := newConfigWithFraction(0.0)
+	config := newTaskQueueConfig(
+		tqid.UnsafeTaskQueueFamily("test-ns", "test-tq").TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW),
+		cfg, "test-ns",
+	)
+	rlm := newRateLimitManager(&mockUserDataManager{}, config, enumspb.TASK_QUEUE_TYPE_WORKFLOW)
+	defer rlm.Stop()
+
+	rlm.SetAPIConfigRPSForTesting(100.0)
+
+	rps, source := rlm.GetEffectiveRPSAndSource()
+	s.InDelta(0.0, rps, 0.0)
+	s.Equal(enumspb.RATE_LIMIT_SOURCE_API, source)
+}