Add caller rate-limit interceptor interface with no-op default

common/rpc/interceptor/caller_rate_limit.go

@@ -0,0 +1,36 @@
+package interceptor
+
+import (
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/common/headers"
+	"go.temporal.io/server/common/namespace"
+)
+
+var (
+	ErrCallerRateLimitExceeded = &serviceerror.ResourceExhausted{
+		Cause:   enumspb.RESOURCE_EXHAUSTED_CAUSE_CALLER_RPS_LIMIT,
+		Scope:   enumspb.RESOURCE_EXHAUSTED_SCOPE_NAMESPACE,
+		Message: "caller rate limit exceeded",
+	}
+)
+
+type (
+	// CallerRateLimitInterceptor rate-limits incoming Nexus requests by caller.
+	// OSS ships only the no-op; Cloud injects the enforcing implementation via fx.
+	CallerRateLimitInterceptor interface {
+		Allow(handlerNamespace *namespace.Namespace, apiName string, headerGetter headers.HeaderGetter) error
+	}
+
+	noopCallerRateLimitInterceptor struct{}
+)
+
+var _ CallerRateLimitInterceptor = (*noopCallerRateLimitInterceptor)(nil)
+
+func NewNoopCallerRateLimitInterceptor() CallerRateLimitInterceptor {
+	return &noopCallerRateLimitInterceptor{}
+}
+
+func (*noopCallerRateLimitInterceptor) Allow(*namespace.Namespace, string, headers.HeaderGetter) error {
+	return nil
+}

common/rpc/interceptor/caller_rate_limit_test.go

@@ -0,0 +1,21 @@
+package interceptor
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/server/common/headers"
+)
+
+func TestErrCallerRateLimitExceeded_CauseAndScope(t *testing.T) {
+	require.Equal(t, enumspb.RESOURCE_EXHAUSTED_CAUSE_CALLER_RPS_LIMIT, ErrCallerRateLimitExceeded.Cause)
+	require.Equal(t, enumspb.RESOURCE_EXHAUSTED_SCOPE_NAMESPACE, ErrCallerRateLimitExceeded.Scope)
+	require.Equal(t, "caller rate limit exceeded", ErrCallerRateLimitExceeded.Message)
+}
+
+func TestNoopCallerRateLimitInterceptor_AllowReturnsNil(t *testing.T) {
+	i := NewNoopCallerRateLimitInterceptor()
+	require.NoError(t, i.Allow(nil, "SomeAPI", headers.NewGRPCHeaderGetter(context.Background())))
+}

service/frontend/fx.go

@@ -94,6 +94,7 @@ var Module = fx.Options(
 	fx.Provide(NamespaceCountLimitInterceptorProvider),
 	fx.Provide(NamespaceValidatorInterceptorProvider),
 	fx.Provide(NamespaceRateLimitInterceptorProvider),
+	fx.Provide(interceptor.NewNoopCallerRateLimitInterceptor),
 	fx.Provide(SDKVersionInterceptorProvider),
 	fx.Provide(CallerInfoInterceptorProvider),
 	fx.Provide(SlowRequestLoggerInterceptorProvider),

service/frontend/nexus_handler.go

@@ -58,6 +58,7 @@ type nexusContext struct {
 	claims                               *authorization.Claims
 	namespaceValidationInterceptor       *interceptor.NamespaceValidatorInterceptor
 	namespaceRateLimitInterceptor        interceptor.NamespaceRateLimitInterceptor
+	callerRateLimitInterceptor           interceptor.CallerRateLimitInterceptor
 	namespaceConcurrencyLimitInterceptor *interceptor.ConcurrentRequestLimitInterceptor
 	rateLimitInterceptor                 *interceptor.RateLimitInterceptor
 	responseHeaders                      map[string]string
@@ -222,6 +223,11 @@ func (c *operationContext) interceptRequest(
 		}
 	})
 
+	if err := c.callerRateLimitInterceptor.Allow(c.namespace, c.apiName, header); err != nil {
+		c.metricsHandler = c.metricsHandler.WithTags(metrics.OutcomeTag("caller_rate_limited"))
+		return commonnexus.ConvertGRPCError(err, true)
+	}
+
 	cleanup, err := c.namespaceConcurrencyLimitInterceptor.Allow(
 		c.namespace.Name(),
 		c.apiName,

service/frontend/nexus_handler_test.go

@@ -75,11 +75,25 @@ type contextOptions struct {
 	namespacePassive        bool
 	quota                   int
 	namespaceRateLimitAllow bool
+	callerRateLimitDeny     bool
 	rateLimitAllow          bool
 	redirectAllow           bool
 	headersBlacklist        []string
 }
 
+type fakeCallerRateLimitInterceptor struct {
+	deny bool
+}
+
+func (f fakeCallerRateLimitInterceptor) Allow(*namespace.Namespace, string, headers.HeaderGetter) error {
+	if f.deny {
+		return interceptor.ErrCallerRateLimitExceeded
+	}
+	return nil
+}
+
+var _ interceptor.CallerRateLimitInterceptor = fakeCallerRateLimitInterceptor{}
+
 func newOperationContext(options contextOptions) *operationContext {
 	oc := &operationContext{
 		nexusContext: &nexusContext{},
@@ -150,6 +164,7 @@ func newOperationContext(options contextOptions) *operationContext {
 		dynamicconfig.GetBoolPropertyFnFilteredByNamespace(false),
 		metrics.NoopMetricsHandler,
 	)
+	oc.callerRateLimitInterceptor = fakeCallerRateLimitInterceptor{deny: options.callerRateLimitDeny}
 	oc.rateLimitInterceptor = interceptor.NewRateLimitInterceptor(
 		mockRateLimiter{options.rateLimitAllow},
 		make(map[string]int),
@@ -253,6 +268,145 @@ func TestNexusInterceptRequest_NamespaceRateLimited_ResultsInResourceExhausted(t
 	require.Equal(t, map[string]string{"outcome": "namespace_rate_limited"}, snap["test"][0].Tags)
 }
 
+func TestNexusInterceptRequest_CallerRateLimited_ResultsInResourceExhausted(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	var err error
+	oc := newOperationContext(contextOptions{
+		namespaceState:          enumspb.NAMESPACE_STATE_REGISTERED,
+		quota:                   1,
+		namespaceRateLimitAllow: true,
+		callerRateLimitDeny:     true,
+		rateLimitAllow:          true,
+	})
+	err = oc.interceptRequest(ctx, &matchingservice.DispatchNexusTaskRequest{}, nexus.Header{})
+	var handlerError *nexus.HandlerError
+	require.ErrorAs(t, err, &handlerError)
+	require.Equal(t, nexus.HandlerErrorTypeResourceExhausted, handlerError.Type)
+	require.Equal(t, "caller rate limit exceeded", handlerError.Message)
+	mh := oc.metricsHandler.(*metricstest.CaptureHandler) //nolint:revive
+	capture := mh.StartCapture()
+	oc.metricsHandler.Counter("test").Record(1)
+	mh.StopCapture(capture)
+	snap := capture.Snapshot()
+	require.Len(t, snap["test"], 1)
+	require.Equal(t, map[string]string{"outcome": "caller_rate_limited"}, snap["test"][0].Tags)
+}
+
+func TestNexusInterceptRequest_CallerRateLimitAllowed_Proceeds(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	oc := newOperationContext(contextOptions{
+		namespaceState:          enumspb.NAMESPACE_STATE_REGISTERED,
+		quota:                   1,
+		namespaceRateLimitAllow: true,
+		callerRateLimitDeny:     false,
+		rateLimitAllow:          true,
+	})
+	err := oc.interceptRequest(ctx, &matchingservice.DispatchNexusTaskRequest{}, nexus.Header{})
+	require.NoError(t, err)
+}
+
+func TestNexusInterceptRequest_CallerRateLimitPrecedesNamespaceRateLimit(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	var err error
+	oc := newOperationContext(contextOptions{
+		namespaceState:          enumspb.NAMESPACE_STATE_REGISTERED,
+		quota:                   1,
+		namespaceRateLimitAllow: false,
+		callerRateLimitDeny:     true,
+		rateLimitAllow:          true,
+	})
+	err = oc.interceptRequest(ctx, &matchingservice.DispatchNexusTaskRequest{}, nexus.Header{})
+	var handlerError *nexus.HandlerError
+	require.ErrorAs(t, err, &handlerError)
+	require.Equal(t, nexus.HandlerErrorTypeResourceExhausted, handlerError.Type)
+	require.Equal(t, "caller rate limit exceeded", handlerError.Message)
+	mh := oc.metricsHandler.(*metricstest.CaptureHandler) //nolint:revive
+	capture := mh.StartCapture()
+	oc.metricsHandler.Counter("test").Record(1)
+	mh.StopCapture(capture)
+	snap := capture.Snapshot()
+	require.Len(t, snap["test"], 1)
+	require.Equal(t, map[string]string{"outcome": "caller_rate_limited"}, snap["test"][0].Tags)
+}
+
+func TestNexusInterceptRequest_CallerRateLimitPrecedesNamespaceConcurrency(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	var err error
+	oc := newOperationContext(contextOptions{
+		namespaceState:          enumspb.NAMESPACE_STATE_REGISTERED,
+		quota:                   0,
+		namespaceRateLimitAllow: true,
+		callerRateLimitDeny:     true,
+		rateLimitAllow:          true,
+	})
+	err = oc.interceptRequest(ctx, &matchingservice.DispatchNexusTaskRequest{}, nexus.Header{})
+	var handlerError *nexus.HandlerError
+	require.ErrorAs(t, err, &handlerError)
+	require.Equal(t, nexus.HandlerErrorTypeResourceExhausted, handlerError.Type)
+	require.Equal(t, "caller rate limit exceeded", handlerError.Message)
+	mh := oc.metricsHandler.(*metricstest.CaptureHandler) //nolint:revive
+	capture := mh.StartCapture()
+	oc.metricsHandler.Counter("test").Record(1)
+	mh.StopCapture(capture)
+	snap := capture.Snapshot()
+	require.Len(t, snap["test"], 1)
+	require.Equal(t, map[string]string{"outcome": "caller_rate_limited"}, snap["test"][0].Tags)
+}
+
+func TestNexusInterceptRequest_CallerRateLimitPrecedesAllOtherLimiters(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	var err error
+	oc := newOperationContext(contextOptions{
+		namespaceState:          enumspb.NAMESPACE_STATE_REGISTERED,
+		quota:                   0,
+		namespaceRateLimitAllow: false,
+		callerRateLimitDeny:     true,
+		rateLimitAllow:          false,
+	})
+	err = oc.interceptRequest(ctx, &matchingservice.DispatchNexusTaskRequest{}, nexus.Header{})
+	var handlerError *nexus.HandlerError
+	require.ErrorAs(t, err, &handlerError)
+	require.Equal(t, nexus.HandlerErrorTypeResourceExhausted, handlerError.Type)
+	require.Equal(t, "caller rate limit exceeded", handlerError.Message)
+	mh := oc.metricsHandler.(*metricstest.CaptureHandler) //nolint:revive
+	capture := mh.StartCapture()
+	oc.metricsHandler.Counter("test").Record(1)
+	mh.StopCapture(capture)
+	snap := capture.Snapshot()
+	require.Len(t, snap["test"], 1)
+	require.Equal(t, map[string]string{"outcome": "caller_rate_limited"}, snap["test"][0].Tags)
+}
+
+func TestNexusInterceptRequest_CallerRateLimitPrecedesGlobal(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	var err error
+	oc := newOperationContext(contextOptions{
+		namespaceState:          enumspb.NAMESPACE_STATE_REGISTERED,
+		quota:                   1,
+		namespaceRateLimitAllow: true,
+		callerRateLimitDeny:     true,
+		rateLimitAllow:          false,
+	})
+	err = oc.interceptRequest(ctx, &matchingservice.DispatchNexusTaskRequest{}, nexus.Header{})
+	var handlerError *nexus.HandlerError
+	require.ErrorAs(t, err, &handlerError)
+	require.Equal(t, nexus.HandlerErrorTypeResourceExhausted, handlerError.Type)
+	require.Equal(t, "caller rate limit exceeded", handlerError.Message)
+	mh := oc.metricsHandler.(*metricstest.CaptureHandler) //nolint:revive
+	capture := mh.StartCapture()
+	oc.metricsHandler.Counter("test").Record(1)
+	mh.StopCapture(capture)
+	snap := capture.Snapshot()
+	require.Len(t, snap["test"], 1)
+	require.Equal(t, map[string]string{"outcome": "caller_rate_limited"}, snap["test"][0].Tags)
+}
+
 func TestNexusInterceptRequest_GlobalRateLimited_ResultsInResourceExhausted(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()

service/frontend/nexus_operation_http_handler.go

@@ -42,6 +42,7 @@ type NexusOperationHTTPHandler struct {
 	auth                                 *authorization.Interceptor
 	namespaceValidationInterceptor       *interceptor.NamespaceValidatorInterceptor
 	namespaceRateLimitInterceptor        interceptor.NamespaceRateLimitInterceptor
+	callerRateLimitInterceptor           interceptor.CallerRateLimitInterceptor
 	namespaceConcurrencyLimitInterceptor *interceptor.ConcurrentRequestLimitInterceptor
 	rateLimitInterceptor                 *interceptor.RateLimitInterceptor
 }
@@ -60,6 +61,7 @@ func NewNexusOperationHTTPHandler(
 	redirectionInterceptor *interceptor.Redirection,
 	namespaceValidationInterceptor *interceptor.NamespaceValidatorInterceptor,
 	namespaceRateLimitInterceptor interceptor.NamespaceRateLimitInterceptor,
+	callerRateLimitInterceptor interceptor.CallerRateLimitInterceptor,
 	namespaceConcurrencyLimitInterceptor *interceptor.ConcurrentRequestLimitInterceptor,
 	rateLimitInterceptor *interceptor.RateLimitInterceptor,
 	logger log.Logger,
@@ -76,6 +78,7 @@ func NewNexusOperationHTTPHandler(
 		auth:                                 authInterceptor,
 		namespaceValidationInterceptor:       namespaceValidationInterceptor,
 		namespaceRateLimitInterceptor:        namespaceRateLimitInterceptor,
+		callerRateLimitInterceptor:           callerRateLimitInterceptor,
 		namespaceConcurrencyLimitInterceptor: namespaceConcurrencyLimitInterceptor,
 		rateLimitInterceptor:                 rateLimitInterceptor,
 		preprocessErrorCounter:               metricsHandler.Counter(metrics.NexusRequestPreProcessErrors.Name()).Record,
@@ -221,6 +224,7 @@ func (h *NexusOperationHTTPHandler) baseNexusContext(apiName string, header http
 	return &nexusContext{
 		namespaceValidationInterceptor:       h.namespaceValidationInterceptor,
 		namespaceRateLimitInterceptor:        h.namespaceRateLimitInterceptor,
+		callerRateLimitInterceptor:           h.callerRateLimitInterceptor,
 		namespaceConcurrencyLimitInterceptor: h.namespaceConcurrencyLimitInterceptor,
 		rateLimitInterceptor:                 h.rateLimitInterceptor,
 		apiName:                              apiName,