From ee7711a4e5367b0404c05768e01bce9d04a08968 Mon Sep 17 00:00:00 2001 From: Prathyush PV Date: Mon, 20 Jul 2026 14:55:09 -0700 Subject: [PATCH] Add caller rate-limit interceptor interface with no-op default --- common/rpc/interceptor/caller_rate_limit.go | 36 ++++ .../rpc/interceptor/caller_rate_limit_test.go | 21 +++ service/frontend/fx.go | 1 + service/frontend/nexus_handler.go | 6 + service/frontend/nexus_handler_test.go | 154 ++++++++++++++++++ .../frontend/nexus_operation_http_handler.go | 4 + 6 files changed, 222 insertions(+) create mode 100644 common/rpc/interceptor/caller_rate_limit.go create mode 100644 common/rpc/interceptor/caller_rate_limit_test.go diff --git a/common/rpc/interceptor/caller_rate_limit.go b/common/rpc/interceptor/caller_rate_limit.go new file mode 100644 index 00000000000..c20c0f8c3c1 --- /dev/null +++ b/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 +} diff --git a/common/rpc/interceptor/caller_rate_limit_test.go b/common/rpc/interceptor/caller_rate_limit_test.go new file mode 100644 index 00000000000..174c37c14d6 --- /dev/null +++ b/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()))) +} diff --git a/service/frontend/fx.go b/service/frontend/fx.go index c771dd300bb..0176137c18a 100644 --- a/service/frontend/fx.go +++ b/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), diff --git a/service/frontend/nexus_handler.go b/service/frontend/nexus_handler.go index 89ff09fe175..9153832a0b9 100644 --- a/service/frontend/nexus_handler.go +++ b/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, diff --git a/service/frontend/nexus_handler_test.go b/service/frontend/nexus_handler_test.go index 0f4d13565c0..a3481204f14 100644 --- a/service/frontend/nexus_handler_test.go +++ b/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() diff --git a/service/frontend/nexus_operation_http_handler.go b/service/frontend/nexus_operation_http_handler.go index e47ef3825ef..7ed9bcbc83c 100644 --- a/service/frontend/nexus_operation_http_handler.go +++ b/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,