diff --git a/chasm/lib/nexusoperation/validator.go b/chasm/lib/nexusoperation/validator.go index de2e58e6a96..adbbe7857ed 100644 --- a/chasm/lib/nexusoperation/validator.go +++ b/chasm/lib/nexusoperation/validator.go @@ -343,6 +343,42 @@ func (v *validator) normalizeIDPolicies(req *workflowservice.StartNexusOperation } } +// ValidateAndNormalizeSyncRequest checks an ExecuteSyncNexusOperation request against the limits +// of standalone operations. It sets a missing request ID and lowercases the Nexus headers in place. +func ValidateAndNormalizeSyncRequest( + config *Config, + logger log.Logger, + req *workflowservice.ExecuteSyncNexusOperationRequest, +) error { + v := &validator{config: config, logger: logger} + if err := v.normalizeRequestID(&req.RequestId); err != nil { + return err + } + if err := v.validateIDLength("identity", req.GetIdentity()); err != nil { + return err + } + // The target and input checks take a start request, and these fields are the same. + start := &workflowservice.StartNexusOperationExecutionRequest{ + Namespace: req.GetNamespace(), + Endpoint: req.GetEndpoint(), + Service: req.GetService(), + Operation: req.GetOperation(), + Input: req.GetInput(), + } + if err := v.validateTarget(start); err != nil { + return err + } + if err := v.validateInput(start); err != nil { + return err + } + loweredHeaders, err := v.validateAndLowercaseHeaders(req.GetNamespace(), req.GetNexusHeader()) + if err != nil { + return err + } + req.NexusHeader = loweredHeaders + return nil +} + // validateOnConflictOptions validates the on_conflict_options of a start request: // - attach_completion_callbacks requires attach_request_id, since it is embedded into the keys we use // to persist them on the Operation's Callbacks map. diff --git a/client/frontend/client_gen.go b/client/frontend/client_gen.go index 2e97d026558..8d003efd61a 100644 --- a/client/frontend/client_gen.go +++ b/client/frontend/client_gen.go @@ -339,6 +339,16 @@ func (c *clientImpl) ExecuteMultiOperation( return c.client.ExecuteMultiOperation(ctx, request, opts...) } +func (c *clientImpl) ExecuteSyncNexusOperation( + ctx context.Context, + request *workflowservice.ExecuteSyncNexusOperationRequest, + opts ...grpc.CallOption, +) (*workflowservice.ExecuteSyncNexusOperationResponse, error) { + ctx, cancel := c.createContext(ctx) + defer cancel() + return c.client.ExecuteSyncNexusOperation(ctx, request, opts...) +} + func (c *clientImpl) FetchWorkerConfig( ctx context.Context, request *workflowservice.FetchWorkerConfigRequest, diff --git a/client/frontend/metric_client_gen.go b/client/frontend/metric_client_gen.go index 04376599642..d81e07a727f 100644 --- a/client/frontend/metric_client_gen.go +++ b/client/frontend/metric_client_gen.go @@ -471,6 +471,20 @@ func (c *metricClient) ExecuteMultiOperation( return c.client.ExecuteMultiOperation(ctx, request, opts...) } +func (c *metricClient) ExecuteSyncNexusOperation( + ctx context.Context, + request *workflowservice.ExecuteSyncNexusOperationRequest, + opts ...grpc.CallOption, +) (_ *workflowservice.ExecuteSyncNexusOperationResponse, retError error) { + + metricsHandler, startTime := c.startMetricsRecording(ctx, "FrontendClientExecuteSyncNexusOperation") + defer func() { + c.finishMetricsRecording(metricsHandler, startTime, retError) + }() + + return c.client.ExecuteSyncNexusOperation(ctx, request, opts...) +} + func (c *metricClient) FetchWorkerConfig( ctx context.Context, request *workflowservice.FetchWorkerConfigRequest, diff --git a/client/frontend/retryable_client_gen.go b/client/frontend/retryable_client_gen.go index 30bce7b47fd..895c38316e3 100644 --- a/client/frontend/retryable_client_gen.go +++ b/client/frontend/retryable_client_gen.go @@ -506,6 +506,21 @@ func (c *retryableClient) ExecuteMultiOperation( return resp, err } +func (c *retryableClient) ExecuteSyncNexusOperation( + ctx context.Context, + request *workflowservice.ExecuteSyncNexusOperationRequest, + opts ...grpc.CallOption, +) (*workflowservice.ExecuteSyncNexusOperationResponse, error) { + var resp *workflowservice.ExecuteSyncNexusOperationResponse + op := func(ctx context.Context) error { + var err error + resp, err = c.client.ExecuteSyncNexusOperation(ctx, request, opts...) + return err + } + err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable) + return resp, err +} + func (c *retryableClient) FetchWorkerConfig( ctx context.Context, request *workflowservice.FetchWorkerConfigRequest, diff --git a/common/api/metadata.go b/common/api/metadata.go index 0a4846c1d0d..b1e66f22553 100644 --- a/common/api/metadata.go +++ b/common/api/metadata.go @@ -109,6 +109,7 @@ var ( "RequestCancelNexusOperationExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, "StartActivityExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, "StartNexusOperationExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, + "ExecuteSyncNexusOperation": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, "TerminateActivityExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, "TerminateNexusOperationExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, "PollNexusTaskQueue": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingAlways}, diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go index 63ef61339d9..8e9961f8eb1 100644 --- a/common/dynamicconfig/constants.go +++ b/common/dynamicconfig/constants.go @@ -1188,6 +1188,13 @@ so forwarding by endpoint ID will not work out of the box.`, `FrontendEnableBatchOperationsForStandaloneActivities controls whether the frontend accepts the batch cancel, terminate, and delete standalone activity operation fields`, ) + FrontendEnableExecuteSyncNexusOperation = NewNamespaceBoolSetting( + "frontend.enableExecuteSyncNexusOperation", + false, + `FrontendEnableExecuteSyncNexusOperation enables the ExecuteSyncNexusOperation API in the +frontend for callers in the namespace.`, + ) + FrontendEnableUpdateWorkflowExecution = NewNamespaceBoolSetting( "frontend.enableUpdateWorkflowExecution", true, diff --git a/common/rpc/interceptor/logtags/workflow_service_server_gen.go b/common/rpc/interceptor/logtags/workflow_service_server_gen.go index b4da09754c7..597dd614f03 100644 --- a/common/rpc/interceptor/logtags/workflow_service_server_gen.go +++ b/common/rpc/interceptor/logtags/workflow_service_server_gen.go @@ -163,6 +163,10 @@ func (wt *WorkflowTags) extractFromWorkflowServiceServerMessage(message any) []t return nil case *workflowservice.ExecuteMultiOperationResponse: return nil + case *workflowservice.ExecuteSyncNexusOperationRequest: + return nil + case *workflowservice.ExecuteSyncNexusOperationResponse: + return nil case *workflowservice.FetchWorkerConfigRequest: return nil case *workflowservice.FetchWorkerConfigResponse: diff --git a/common/rpc/interceptor/redirection.go b/common/rpc/interceptor/redirection.go index a7dedd5e1ac..39c763f2826 100644 --- a/common/rpc/interceptor/redirection.go +++ b/common/rpc/interceptor/redirection.go @@ -43,6 +43,10 @@ var ( "GetSearchAttributes": func() any { return &workflowservice.GetSearchAttributesResponse{} }, "GetClusterInfo": func() any { return &workflowservice.GetClusterInfoResponse{} }, "GetSystemInfo": func() any { return &workflowservice.GetSystemInfoResponse{} }, + + // The handler checks the target namespace, which can be active in another cluster than the + // caller namespace. + "ExecuteSyncNexusOperation": func() any { return &workflowservice.ExecuteSyncNexusOperationResponse{} }, } globalAPIResponses = map[string]responseConstructorFn{ diff --git a/common/rpc/interceptor/redirection_test.go b/common/rpc/interceptor/redirection_test.go index 49524fd2394..cb6a2284d12 100644 --- a/common/rpc/interceptor/redirection_test.go +++ b/common/rpc/interceptor/redirection_test.go @@ -95,6 +95,8 @@ func (s *redirectionInterceptorSuite) TestLocalAPI() { "GetSearchAttributes": {}, "GetClusterInfo": {}, "GetSystemInfo": {}, + + "ExecuteSyncNexusOperation": {}, }, apis) } diff --git a/common/testing/mockapi/workflowservicemock/v1/service_grpc.pb.mock.go b/common/testing/mockapi/workflowservicemock/v1/service_grpc.pb.mock.go index 619a71d7a0a..499ee2f7918 100644 --- a/common/testing/mockapi/workflowservicemock/v1/service_grpc.pb.mock.go +++ b/common/testing/mockapi/workflowservicemock/v1/service_grpc.pb.mock.go @@ -702,6 +702,26 @@ func (mr *MockWorkflowServiceClientMockRecorder) ExecuteMultiOperation(ctx, in a return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ExecuteMultiOperation", reflect.TypeOf((*MockWorkflowServiceClient)(nil).ExecuteMultiOperation), varargs...) } +// ExecuteSyncNexusOperation mocks base method. +func (m *MockWorkflowServiceClient) ExecuteSyncNexusOperation(ctx context.Context, in *workflowservice.ExecuteSyncNexusOperationRequest, opts ...grpc.CallOption) (*workflowservice.ExecuteSyncNexusOperationResponse, error) { + m.ctrl.T.Helper() + varargs := []any{ctx, in} + for _, a := range opts { + varargs = append(varargs, a) + } + ret := m.ctrl.Call(m, "ExecuteSyncNexusOperation", varargs...) + ret0, _ := ret[0].(*workflowservice.ExecuteSyncNexusOperationResponse) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ExecuteSyncNexusOperation indicates an expected call of ExecuteSyncNexusOperation. +func (mr *MockWorkflowServiceClientMockRecorder) ExecuteSyncNexusOperation(ctx, in any, opts ...any) *gomock.Call { + mr.mock.ctrl.T.Helper() + varargs := append([]any{ctx, in}, opts...) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ExecuteSyncNexusOperation", reflect.TypeOf((*MockWorkflowServiceClient)(nil).ExecuteSyncNexusOperation), varargs...) +} + // FetchWorkerConfig mocks base method. func (m *MockWorkflowServiceClient) FetchWorkerConfig(ctx context.Context, in *workflowservice.FetchWorkerConfigRequest, opts ...grpc.CallOption) (*workflowservice.FetchWorkerConfigResponse, error) { m.ctrl.T.Helper() diff --git a/go.mod b/go.mod index 593ce68d4c4..2f83b4a3241 100644 --- a/go.mod +++ b/go.mod @@ -224,4 +224,4 @@ require ( tool golang.org/x/perf/cmd/benchstat -replace go.temporal.io/api => github.com/moetemp/api-go v1.63.6-0.20261016104405-5d5af69e7748 +replace go.temporal.io/api => github.com/moetemp/api-go v1.63.6-0.20261016104423-a090bbf0413c diff --git a/go.sum b/go.sum index a0f0fe29167..16f3dec9257 100644 --- a/go.sum +++ b/go.sum @@ -319,8 +319,8 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8= github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= -github.com/moetemp/api-go v1.63.6-0.20261016104405-5d5af69e7748 h1:YMY5nOoBhv17FNekWZj5Co4ktqSoEKgConNior8/prg= -github.com/moetemp/api-go v1.63.6-0.20261016104405-5d5af69e7748/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= +github.com/moetemp/api-go v1.63.6-0.20261016104423-a090bbf0413c h1:gwxe6Y8xFv+z6WVGvR0pw8IM+UD8RVtmUl4qyqSFvXY= +github.com/moetemp/api-go v1.63.6-0.20261016104423-a090bbf0413c/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= diff --git a/service/frontend/configs/quotas.go b/service/frontend/configs/quotas.go index 0d7d343adbb..a8730dc887f 100644 --- a/service/frontend/configs/quotas.go +++ b/service/frontend/configs/quotas.go @@ -63,6 +63,9 @@ var ( // Dispatching a Nexus task is a potentially long running RPC, it's classified in the same bucket as QueryWorkflow. DispatchNexusTaskByNamespaceAndTaskQueueAPIName: 1, DispatchNexusTaskByEndpointAPIName: 1, + + // A sync Nexus call waits for its Nexus task to finish, as a dispatch does. + "/temporal.api.workflowservice.v1.WorkflowService/ExecuteSyncNexusOperation": 1, } // PollTaskAPISet is the set of API methods for which NamespaceRateLimitInterceptor will @@ -96,6 +99,7 @@ var ( "/temporal.api.workflowservice.v1.WorkflowService/StartBatchOperation": 1, "/temporal.api.workflowservice.v1.WorkflowService/StartActivityExecution": 1, "/temporal.api.workflowservice.v1.WorkflowService/StartNexusOperationExecution": 1, + "/temporal.api.workflowservice.v1.WorkflowService/ExecuteSyncNexusOperation": 1, "/temporal.api.workflowservice.v1.WorkflowService/AttachStreamCallback": 1, DispatchNexusTaskByNamespaceAndTaskQueueAPIName: 1, DispatchNexusTaskByEndpointAPIName: 1, diff --git a/service/frontend/fx.go b/service/frontend/fx.go index 342fe023dc0..f9d0c2b093c 100644 --- a/service/frontend/fx.go +++ b/service/frontend/fx.go @@ -33,6 +33,7 @@ import ( "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/namespace" "go.temporal.io/server/common/namespace/nsreplication" + commonnexus "go.temporal.io/server/common/nexus" "go.temporal.io/server/common/persistence" "go.temporal.io/server/common/persistence/serialization" "go.temporal.io/server/common/persistence/visibility" @@ -1003,6 +1004,12 @@ func HandlerProvider( streamNotifierHandler streamnotifier.FrontendHandler, registry *chasm.Registry, frontendServiceResolver membership.ServiceResolver, + endpointRegistry commonnexus.EndpointRegistry, + namespaceRateLimiter interceptor.NamespaceRateLimitInterceptor, + nexusOperationConfig *chasmnexus.Config, + authInterceptor *authorization.Interceptor, + namespaceValidator *interceptor.NamespaceValidatorInterceptor, + namespaceCountLimiter *interceptor.ConcurrentRequestLimitInterceptor, ) Handler { workerDeploymentReadRateLimiter := configs.NewGlobalNamespaceRateLimiter( frontendServiceResolver, @@ -1050,6 +1057,14 @@ func HandlerProvider( saMapperProvider, saValidator, ), + endpointRegistry, + namespaceRateLimiter, + nexusOperationConfig, + SyncNexusChecks{ + Authorizer: authInterceptor, + StateChecker: namespaceValidator, + CountLimiter: namespaceCountLimiter, + }, ) return wfHandler } diff --git a/service/frontend/nexus_sync.go b/service/frontend/nexus_sync.go new file mode 100644 index 00000000000..a22fa01d5be --- /dev/null +++ b/service/frontend/nexus_sync.go @@ -0,0 +1,202 @@ +package frontend + +import ( + "context" + "time" + + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + nexuspb "go.temporal.io/api/nexus/v1" + "go.temporal.io/api/serviceerror" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/server/api/matchingservice/v1" + chasmnexus "go.temporal.io/server/chasm/lib/nexusoperation" + "go.temporal.io/server/common/api" + "go.temporal.io/server/common/authorization" + "go.temporal.io/server/common/headers" + "go.temporal.io/server/common/log" + "go.temporal.io/server/common/namespace" + commonnexus "go.temporal.io/server/common/nexus" + "go.temporal.io/server/service/frontend/configs" + "google.golang.org/protobuf/types/known/timestamppb" +) + +var ( + errExecuteSyncNexusOperationDisabled = serviceerror.NewUnimplemented( + "ExecuteSyncNexusOperation is disabled for this namespace") + errSyncNexusOperationAnsweredAsync = serviceerror.NewFailedPrecondition( + "the Nexus operation answered asynchronously. " + + "Use StartNexusOperationExecution for asynchronous operations") + errSyncNexusOperationNoAnswer = serviceerror.NewDeadlineExceeded( + "no worker answered the Nexus task before the deadline") +) + +// ExecuteSyncNexusOperation runs one synchronous Nexus operation for a caller in the request +// namespace. The gRPC interceptors already authorized the caller and counted its rate limit +// against the request namespace. Nothing is persisted, and the caller owns the retries. +func (wh *WorkflowHandler) ExecuteSyncNexusOperation( + ctx context.Context, + request *workflowservice.ExecuteSyncNexusOperationRequest, +) (_ *workflowservice.ExecuteSyncNexusOperationResponse, retError error) { + defer log.CapturePanic(wh.logger, &retError) + + if request == nil { + return nil, errRequestNotSet + } + if request.GetNamespace() == "" { + return nil, errNamespaceNotSet + } + if !wh.config.EnableExecuteSyncNexusOperation(request.GetNamespace()) { + return nil, errExecuteSyncNexusOperationDisabled + } + + callerNamespace, err := wh.namespaceRegistry.GetNamespace( + namespace.Name(request.GetNamespace()), + ) + if err != nil { + return nil, err + } + err = chasmnexus.ValidateAndNormalizeSyncRequest(wh.nexusOperationConfig, wh.logger, request) + if err != nil { + return nil, err + } + + // The gRPC interceptor authorized the caller namespace without the endpoint. An authorizer that + // limits endpoints by name on the Nexus HTTP route gets the name here too. + claims, _ := ctx.Value(authorization.MappedClaims).(*authorization.Claims) + if _, err := wh.syncNexusChecks.Authorizer.Authorize(ctx, claims, &authorization.CallTarget{ + APIName: api.WorkflowServicePrefix + "ExecuteSyncNexusOperation", + Namespace: callerNamespace.Name().String(), + NexusEndpointName: request.GetEndpoint(), + Request: request, + }); err != nil { + return nil, err + } + + // The registry is the allowlist seam. It answers NotFound for an endpoint that the caller + // namespace can't reach, so a caller can't probe endpoint names. + entry, err := wh.endpointRegistry.GetByName(ctx, callerNamespace.ID(), request.GetEndpoint()) + if err != nil { + return nil, err + } + workerTarget := entry.GetEndpoint().GetSpec().GetTarget().GetWorker() + if workerTarget == nil { + return nil, serviceerror.NewInvalidArgumentf( + "endpoint %q has no worker target. ExecuteSyncNexusOperation calls worker targets only", + request.GetEndpoint()) + } + + targetNamespace, err := wh.namespaceRegistry.GetNamespaceByID( + namespace.ID(workerTarget.GetNamespaceId()), + ) + if err != nil { + return nil, err + } + // A target that is deleted or in a handover takes no task, as on the Nexus HTTP route. + err = wh.syncNexusChecks.StateChecker.ValidateState( + targetNamespace, + configs.DispatchNexusTaskByEndpointAPIName, + namespace.EmptyBusinessID, + ) + if err != nil { + return nil, err + } + currentCluster := wh.clusterMetadata.GetCurrentClusterName() + //nolint:forbidigo // Nexus requests are not tied to a business ID, as on the Nexus HTTP route. + if !targetNamespace.ActiveInCluster(currentCluster) { + return nil, serviceerror.NewNamespaceNotActive( + targetNamespace.Name().String(), + currentCluster, + targetNamespace.ActiveClusterName(namespace.RoutingKey{}), + ) + } + release, err := wh.syncNexusChecks.CountLimiter.Allow( + targetNamespace.Name(), + configs.DispatchNexusTaskByEndpointAPIName, + wh.metricsScope(ctx), + request, + ) + if err != nil { + return nil, err + } + defer release() + // The target namespace pays for the dispatch as it does for a call on the Nexus HTTP route. + if err := wh.namespaceRateLimiter.Allow( + ctx, + targetNamespace.Name(), + configs.DispatchNexusTaskByEndpointAPIName, + headers.NewGRPCHeaderGetter(ctx), + ); err != nil { + return nil, err + } + + // A shorter caller deadline wins. Matching passes the deadline to the handler. + ctx, cancel := context.WithTimeout( + ctx, + wh.nexusOperationConfig.RequestTimeout( + callerNamespace.Name().String(), + request.GetEndpoint(), + ), + ) + defer cancel() + + resp, err := wh.matchingClient.DispatchNexusTask(ctx, &matchingservice.DispatchNexusTaskRequest{ + NamespaceId: targetNamespace.ID().String(), + TaskQueue: &taskqueuepb.TaskQueue{ + Name: workerTarget.GetTaskQueue(), + Kind: enumspb.TASK_QUEUE_KIND_NORMAL, + }, + Request: &nexuspb.Request{ + Header: request.GetNexusHeader(), + ScheduledTime: timestamppb.New(time.Now().UTC()), + Endpoint: request.GetEndpoint(), + Variant: &nexuspb.Request_StartOperation{ + StartOperation: &nexuspb.StartOperationRequest{ + Service: request.GetService(), + Operation: request.GetOperation(), + RequestId: request.GetRequestId(), + Payload: request.GetInput(), + }, + }, + Capabilities: &nexuspb.Request_Capabilities{ + TemporalFailureResponses: true, + }, + }, + }) + if err != nil { + return nil, err + } + return syncNexusOperationResponse(commonnexus.ClassifyStartOperationDispatch(resp)) +} + +func syncNexusOperationResponse( + result commonnexus.DispatchResult, +) (*workflowservice.ExecuteSyncNexusOperationResponse, error) { + switch result.Outcome { + case commonnexus.DispatchOutcomeSyncSuccess: + payload := result.OperationResult + if payload == nil { + // An operation may succeed with no value. The response still says that it succeeded. + payload = &commonpb.Payload{} + } + return &workflowservice.ExecuteSyncNexusOperationResponse{ + Outcome: &workflowservice.ExecuteSyncNexusOperationResponse_Result{Result: payload}, + }, nil + case commonnexus.DispatchOutcomeHandlerFailure, commonnexus.DispatchOutcomeOperationFailure: + return &workflowservice.ExecuteSyncNexusOperationResponse{ + Outcome: &workflowservice.ExecuteSyncNexusOperationResponse_Failure{ + Failure: result.Failure, + }, + }, nil + case commonnexus.DispatchOutcomeAsyncSuccess: + return nil, errSyncNexusOperationAnsweredAsync + case commonnexus.DispatchOutcomeRequestTimeout: + return nil, errSyncNexusOperationNoAnswer + default: + return nil, serviceerror.NewInternalf( + "unrecognized Nexus dispatch outcome: %s", + result.Outcome, + ) + } +} diff --git a/service/frontend/nexus_sync_checks.go b/service/frontend/nexus_sync_checks.go new file mode 100644 index 00000000000..02c0ced1fa4 --- /dev/null +++ b/service/frontend/nexus_sync_checks.go @@ -0,0 +1,33 @@ +package frontend + +import ( + "context" + + commonpb "go.temporal.io/api/common/v1" + "go.temporal.io/server/common/authorization" + "go.temporal.io/server/common/metrics" + "go.temporal.io/server/common/namespace" +) + +// SyncNexusChecks are the checks the Nexus HTTP route runs before a dispatch, so that +// ExecuteSyncNexusOperation admits the same calls. +type SyncNexusChecks struct { + Authorizer interface { + Authorize( + ctx context.Context, + claims *authorization.Claims, + target *authorization.CallTarget, + ) (*commonpb.Principal, error) + } + StateChecker interface { + ValidateState(ns *namespace.Namespace, fullMethod string, businessID string) error + } + CountLimiter interface { + Allow( + ns namespace.Name, + methodName string, + handler metrics.Handler, + req any, + ) (func(), error) + } +} diff --git a/service/frontend/nexus_sync_test.go b/service/frontend/nexus_sync_test.go new file mode 100644 index 00000000000..68c67e4b3c4 --- /dev/null +++ b/service/frontend/nexus_sync_test.go @@ -0,0 +1,572 @@ +package frontend + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + failurepb "go.temporal.io/api/failure/v1" + nexuspb "go.temporal.io/api/nexus/v1" + "go.temporal.io/api/serviceerror" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/server/api/matchingservice/v1" + "go.temporal.io/server/api/matchingservicemock/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" + chasmnexus "go.temporal.io/server/chasm/lib/nexusoperation" + "go.temporal.io/server/common/authorization" + "go.temporal.io/server/common/cluster" + "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/headers" + "go.temporal.io/server/common/log" + "go.temporal.io/server/common/metrics" + "go.temporal.io/server/common/namespace" + "go.temporal.io/server/common/nexus/nexustest" + "go.temporal.io/server/common/rpc/interceptor" + "go.temporal.io/server/service/frontend/configs" + "go.uber.org/mock/gomock" +) + +const ( + syncTestCallerNamespace = "caller-ns" + syncTestCallerID = "caller-ns-id" + syncTestTargetNamespace = "target-ns" + syncTestTargetID = "target-ns-id" + syncTestEndpoint = "stream-endpoint" + syncTestTaskQueue = "stream-tq" +) + +// fakeSyncRateLimiter records the target namespace checks and answers with err. +type fakeSyncRateLimiter struct { + interceptor.NamespaceRateLimitInterceptor + calls []string + err error +} + +func (f *fakeSyncRateLimiter) Allow( + _ context.Context, + namespaceName namespace.Name, + methodName string, + _ headers.HeaderGetter, +) error { + f.calls = append(f.calls, namespaceName.String()+" "+methodName) + return f.err +} + +// fakeSyncAuthorizer records each call target and answers with err. +type fakeSyncAuthorizer struct { + targets []*authorization.CallTarget + err error +} + +func (f *fakeSyncAuthorizer) Authorize( + _ context.Context, + _ *authorization.Claims, + target *authorization.CallTarget, +) (*commonpb.Principal, error) { + f.targets = append(f.targets, target) + return nil, f.err +} + +// fakeSyncCountLimiter records the target namespace checks and answers with err. +type fakeSyncCountLimiter struct { + calls []string + released int + err error +} + +func (f *fakeSyncCountLimiter) Allow( + namespaceName namespace.Name, + methodName string, + _ metrics.Handler, + _ any, +) (func(), error) { + f.calls = append(f.calls, namespaceName.String()+" "+methodName) + return func() { f.released++ }, f.err +} + +type syncTestEnv struct { + handler *WorkflowHandler + matching *matchingservicemock.MockMatchingServiceClient + rateLimiter *fakeSyncRateLimiter + authorizer *fakeSyncAuthorizer + countLimiter *fakeSyncCountLimiter + endpointCalls []namespace.ID +} + +func newSyncTestEnv( + t *testing.T, + enabled bool, + targetActive bool, + targetOpts ...func(*persistencespb.NamespaceInfo), +) *syncTestEnv { + ctrl := gomock.NewController(t) + dc := dynamicconfig.NewNoopCollection() + + activeCluster := cluster.TestCurrentClusterName + if !targetActive { + activeCluster = cluster.TestAlternativeClusterName + } + callerNS := namespace.NewLocalNamespaceForTest( + &persistencespb.NamespaceInfo{Id: syncTestCallerID, Name: syncTestCallerNamespace}, + nil, + cluster.TestCurrentClusterName, + ) + targetInfo := &persistencespb.NamespaceInfo{ + Id: syncTestTargetID, + Name: syncTestTargetNamespace, + State: enumspb.NAMESPACE_STATE_REGISTERED, + } + for _, opt := range targetOpts { + opt(targetInfo) + } + targetNS := namespace.NewGlobalNamespaceForTest( + targetInfo, + nil, + &persistencespb.NamespaceReplicationConfig{ + ActiveClusterName: activeCluster, + Clusters: []string{ + cluster.TestCurrentClusterName, + cluster.TestAlternativeClusterName, + }, + }, + 1, + ) + nsRegistry := namespace.NewMockRegistry(ctrl) + nsRegistry.EXPECT(). + GetNamespace(namespace.Name(syncTestCallerNamespace)). + Return(callerNS, nil). + AnyTimes() + nsRegistry.EXPECT(). + GetNamespaceByID(namespace.ID(syncTestTargetID)). + Return(targetNS, nil). + AnyTimes() + + clusterMetadata := cluster.NewMockMetadata(ctrl) + clusterMetadata.EXPECT(). + GetCurrentClusterName(). + Return(cluster.TestCurrentClusterName). + AnyTimes() + + env := &syncTestEnv{ + matching: matchingservicemock.NewMockMatchingServiceClient(ctrl), + rateLimiter: &fakeSyncRateLimiter{}, + authorizer: &fakeSyncAuthorizer{}, + countLimiter: &fakeSyncCountLimiter{}, + } + registry := nexustest.FakeEndpointRegistry{ + OnGetByName: func( + _ context.Context, + namespaceID namespace.ID, + name string, + ) (*persistencespb.NexusEndpointEntry, error) { + env.endpointCalls = append(env.endpointCalls, namespaceID) + switch name { + case syncTestEndpoint: + return &persistencespb.NexusEndpointEntry{ + Id: "endpoint-id", + Endpoint: &persistencespb.NexusEndpoint{ + Spec: &persistencespb.NexusEndpointSpec{ + Name: name, + Target: &persistencespb.NexusEndpointTarget{ + Variant: &persistencespb.NexusEndpointTarget_Worker_{ + Worker: &persistencespb.NexusEndpointTarget_Worker{ + NamespaceId: syncTestTargetID, + TaskQueue: syncTestTaskQueue, + }, + }, + }, + }, + }, + }, nil + case "external-endpoint": + return &persistencespb.NexusEndpointEntry{ + Id: "external-id", + Endpoint: &persistencespb.NexusEndpoint{ + Spec: &persistencespb.NexusEndpointSpec{ + Name: name, + Target: &persistencespb.NexusEndpointTarget{ + Variant: &persistencespb.NexusEndpointTarget_External_{ + External: &persistencespb.NexusEndpointTarget_External{ + Url: "http://localhost", + }, + }, + }, + }, + }, + }, nil + default: + return nil, serviceerror.NewNotFoundf( + "could not find Nexus endpoint by name: %v", + name, + ) + } + }, + } + + env.handler = &WorkflowHandler{ + config: &Config{ + EnableExecuteSyncNexusOperation: dynamicconfig.GetBoolPropertyFnFilteredByNamespace( + enabled, + ), + }, + logger: log.NewNoopLogger(), + namespaceRegistry: nsRegistry, + clusterMetadata: clusterMetadata, + matchingClient: env.matching, + endpointRegistry: registry, + namespaceRateLimiter: env.rateLimiter, + syncNexusChecks: SyncNexusChecks{ + Authorizer: env.authorizer, + StateChecker: interceptor.NewNamespaceValidatorInterceptor( + nsRegistry, + dynamicconfig.GetBoolPropertyFn(false), + dynamicconfig.GetIntPropertyFn(1000), + nil, + ), + CountLimiter: env.countLimiter, + }, + nexusOperationConfig: &chasmnexus.Config{ + RequestTimeout: chasmnexus.RequestTimeout.Get(dc), + MaxServiceNameLength: chasmnexus.MaxServiceNameLength.Get(dc), + MaxOperationNameLength: chasmnexus.MaxOperationNameLength.Get(dc), + MaxOperationHeaderSize: chasmnexus.MaxOperationHeaderSize.Get(dc), + DisallowedOperationHeaders: chasmnexus.DisallowedOperationHeaders.Get(dc), + PayloadSizeLimit: dynamicconfig.BlobSizeLimitError.Get(dc), + PayloadSizeLimitWarn: dynamicconfig.BlobSizeLimitWarn.Get(dc), + MaxIDLengthLimit: dynamicconfig.MaxIDLengthLimit.Get(dc), + }, + } + return env +} + +func syncTestRequest() *workflowservice.ExecuteSyncNexusOperationRequest { + return &workflowservice.ExecuteSyncNexusOperationRequest{ + Namespace: syncTestCallerNamespace, + RequestId: "request-id", + Endpoint: syncTestEndpoint, + Service: "temporal.sdk.streams.v1.TemporalStreams", + Operation: "read", + Input: &commonpb.Payload{Data: []byte(`{"stream":"s"}`)}, + NexusHeader: map[string]string{"X-Trace": "abc"}, + } +} + +func syncTestDispatchResponse( + start *nexuspb.StartOperationResponse, +) *matchingservice.DispatchNexusTaskResponse { + return &matchingservice.DispatchNexusTaskResponse{ + Outcome: &matchingservice.DispatchNexusTaskResponse_Response{ + Response: &nexuspb.Response{ + Variant: &nexuspb.Response_StartOperation{StartOperation: start}, + }, + }, + } +} + +func TestExecuteSyncNexusOperation_SyncResult(t *testing.T) { + env := newSyncTestEnv(t, true, true) + result := &commonpb.Payload{Data: []byte(`{"records":[]}`)} + + env.matching.EXPECT().DispatchNexusTask(gomock.Any(), gomock.Any()).DoAndReturn( + func( + ctx context.Context, + req *matchingservice.DispatchNexusTaskRequest, + _ ...any, + ) (*matchingservice.DispatchNexusTaskResponse, error) { + deadline, ok := ctx.Deadline() + require.True(t, ok) + require.LessOrEqual(t, time.Until(deadline), 10*time.Second) + + require.Equal(t, syncTestTargetID, req.GetNamespaceId()) + require.Equal(t, syncTestTaskQueue, req.GetTaskQueue().GetName()) + require.Equal(t, syncTestEndpoint, req.GetRequest().GetEndpoint()) + require.Equal(t, map[string]string{"x-trace": "abc"}, req.GetRequest().GetHeader()) + require.True(t, req.GetRequest().GetCapabilities().GetTemporalFailureResponses()) + start := req.GetRequest().GetStartOperation() + require.Equal(t, "temporal.sdk.streams.v1.TemporalStreams", start.GetService()) + require.Equal(t, "read", start.GetOperation()) + require.Equal(t, "request-id", start.GetRequestId()) + require.Empty(t, start.GetCallback()) + require.JSONEq(t, `{"stream":"s"}`, string(start.GetPayload().GetData())) + return syncTestDispatchResponse(&nexuspb.StartOperationResponse{ + Variant: &nexuspb.StartOperationResponse_SyncSuccess{ + SyncSuccess: &nexuspb.StartOperationResponse_Sync{Payload: result}, + }, + }), nil + }) + + resp, err := env.handler.ExecuteSyncNexusOperation(context.Background(), syncTestRequest()) + require.NoError(t, err) + require.Equal(t, result.GetData(), resp.GetResult().GetData()) + require.Equal(t, []namespace.ID{syncTestCallerID}, env.endpointCalls) + require.Equal(t, + []string{syncTestTargetNamespace + " " + configs.DispatchNexusTaskByEndpointAPIName}, + env.rateLimiter.calls) +} + +func TestExecuteSyncNexusOperation_SyncResultWithNoValue(t *testing.T) { + env := newSyncTestEnv(t, true, true) + env.matching.EXPECT().DispatchNexusTask(gomock.Any(), gomock.Any()).Return( + syncTestDispatchResponse(&nexuspb.StartOperationResponse{ + Variant: &nexuspb.StartOperationResponse_SyncSuccess{ + SyncSuccess: &nexuspb.StartOperationResponse_Sync{}, + }, + }), nil) + + resp, err := env.handler.ExecuteSyncNexusOperation(context.Background(), syncTestRequest()) + require.NoError(t, err) + require.NotNil(t, resp.GetResult()) +} + +func TestExecuteSyncNexusOperation_Failures(t *testing.T) { + handlerFailure := &failurepb.Failure{ + Message: "bad stream name", + FailureInfo: &failurepb.Failure_NexusHandlerFailureInfo{ + NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{Type: "BAD_REQUEST"}, + }, + } + operationFailure := &failurepb.Failure{ + Message: "replay mismatch", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{ + Type: "StreamReplayMismatchError", + }, + }, + } + canceledFailure := &failurepb.Failure{ + Message: "canceled", + FailureInfo: &failurepb.Failure_CanceledFailureInfo{ + CanceledFailureInfo: &failurepb.CanceledFailureInfo{}, + }, + } + + cases := []struct { + name string + response *matchingservice.DispatchNexusTaskResponse + want *failurepb.Failure + }{ + { + name: "handler error", + response: &matchingservice.DispatchNexusTaskResponse{ + Outcome: &matchingservice.DispatchNexusTaskResponse_Failure{ + Failure: handlerFailure, + }, + }, + want: handlerFailure, + }, + { + name: "operation failure", + response: syncTestDispatchResponse(&nexuspb.StartOperationResponse{ + Variant: &nexuspb.StartOperationResponse_Failure{Failure: operationFailure}, + }), + want: operationFailure, + }, + { + name: "operation canceled", + response: syncTestDispatchResponse(&nexuspb.StartOperationResponse{ + Variant: &nexuspb.StartOperationResponse_Failure{Failure: canceledFailure}, + }), + want: canceledFailure, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + env := newSyncTestEnv(t, true, true) + env.matching.EXPECT(). + DispatchNexusTask(gomock.Any(), gomock.Any()). + Return(tc.response, nil) + + resp, err := env.handler.ExecuteSyncNexusOperation( + context.Background(), + syncTestRequest(), + ) + require.NoError(t, err) + require.Nil(t, resp.GetResult()) + require.Equal(t, tc.want.GetMessage(), resp.GetFailure().GetMessage()) + require.Equal(t, tc.want.GetFailureInfo(), resp.GetFailure().GetFailureInfo()) + }) + } +} + +func TestExecuteSyncNexusOperation_AsyncAnswerRefused(t *testing.T) { + env := newSyncTestEnv(t, true, true) + env.matching.EXPECT().DispatchNexusTask(gomock.Any(), gomock.Any()).Return( + syncTestDispatchResponse(&nexuspb.StartOperationResponse{ + Variant: &nexuspb.StartOperationResponse_AsyncSuccess{ + AsyncSuccess: &nexuspb.StartOperationResponse_Async{OperationToken: "token"}, + }, + }), nil) + + _, err := env.handler.ExecuteSyncNexusOperation(context.Background(), syncTestRequest()) + var precondition *serviceerror.FailedPrecondition + require.ErrorAs(t, err, &precondition) +} + +func TestExecuteSyncNexusOperation_NoWorkerAnswered(t *testing.T) { + env := newSyncTestEnv(t, true, true) + env.matching.EXPECT().DispatchNexusTask(gomock.Any(), gomock.Any()).Return( + &matchingservice.DispatchNexusTaskResponse{ + Outcome: &matchingservice.DispatchNexusTaskResponse_RequestTimeout{ + RequestTimeout: &matchingservice.DispatchNexusTaskResponse_Timeout{}, + }, + }, nil) + + _, err := env.handler.ExecuteSyncNexusOperation(context.Background(), syncTestRequest()) + var deadline *serviceerror.DeadlineExceeded + require.ErrorAs(t, err, &deadline) +} + +func TestExecuteSyncNexusOperation_RefusedBeforeDispatch(t *testing.T) { + cases := []struct { + name string + enabled bool + targetActive bool + rateLimitErr error + mutate func(*workflowservice.ExecuteSyncNexusOperationRequest) + check func(*testing.T, error) + }{ + { + name: "disabled", + enabled: false, + targetActive: true, + check: func(t *testing.T, err error) { + var unimplemented *serviceerror.Unimplemented + require.ErrorAs(t, err, &unimplemented) + }, + }, + { + name: "unknown endpoint", + enabled: true, + targetActive: true, + mutate: func(req *workflowservice.ExecuteSyncNexusOperationRequest) { + req.Endpoint = "missing" + }, + check: func(t *testing.T, err error) { + var notFound *serviceerror.NotFound + require.ErrorAs(t, err, ¬Found) + }, + }, + { + name: "external endpoint", + enabled: true, + targetActive: true, + mutate: func(req *workflowservice.ExecuteSyncNexusOperationRequest) { + req.Endpoint = "external-endpoint" + }, + check: func(t *testing.T, err error) { + var invalid *serviceerror.InvalidArgument + require.ErrorAs(t, err, &invalid) + }, + }, + { + name: "target namespace not active", + enabled: true, + targetActive: false, + check: func(t *testing.T, err error) { + var notActive *serviceerror.NamespaceNotActive + require.ErrorAs(t, err, ¬Active) + require.Equal(t, cluster.TestAlternativeClusterName, notActive.ActiveCluster) + }, + }, + { + name: "target namespace rate limited", + enabled: true, + targetActive: true, + rateLimitErr: serviceerror.NewResourceExhausted(0, "namespace rate limit exceeded"), + check: func(t *testing.T, err error) { + var exhausted *serviceerror.ResourceExhausted + require.ErrorAs(t, err, &exhausted) + }, + }, + { + name: "input above the blob size limit", + enabled: true, + targetActive: true, + mutate: func(req *workflowservice.ExecuteSyncNexusOperationRequest) { + limitFn := dynamicconfig.BlobSizeLimitError.Get(dynamicconfig.NewNoopCollection()) + limit := limitFn(syncTestCallerNamespace) + req.Input = &commonpb.Payload{Data: make([]byte, limit+1)} + }, + check: func(t *testing.T, err error) { + var invalid *serviceerror.InvalidArgument + require.ErrorAs(t, err, &invalid) + require.Contains(t, invalid.Error(), "input exceeds size limit") + }, + }, + { + name: "missing service", + enabled: true, + targetActive: true, + mutate: func(req *workflowservice.ExecuteSyncNexusOperationRequest) { + req.Service = "" + }, + check: func(t *testing.T, err error) { + var invalid *serviceerror.InvalidArgument + require.ErrorAs(t, err, &invalid) + }, + }, + { + name: "disallowed header", + enabled: true, + targetActive: true, + mutate: func(req *workflowservice.ExecuteSyncNexusOperationRequest) { + req.NexusHeader = map[string]string{"Request-Timeout": "1s"} + }, + check: func(t *testing.T, err error) { + var invalid *serviceerror.InvalidArgument + require.ErrorAs(t, err, &invalid) + }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + env := newSyncTestEnv(t, tc.enabled, tc.targetActive) + env.rateLimiter.err = tc.rateLimitErr + req := syncTestRequest() + if tc.mutate != nil { + tc.mutate(req) + } + // The matching mock fails the test if anything dispatches. + _, err := env.handler.ExecuteSyncNexusOperation(context.Background(), req) + tc.check(t, err) + }) + } +} + +func TestExecuteSyncNexusOperation_AuthorizesTheEndpoint(t *testing.T) { + env := newSyncTestEnv(t, true, true) + env.authorizer.err = serviceerror.NewPermissionDenied("endpoint not allowed", "") + _, err := env.handler.ExecuteSyncNexusOperation(context.Background(), syncTestRequest()) + var denied *serviceerror.PermissionDenied + require.ErrorAs(t, err, &denied) + require.Len(t, env.authorizer.targets, 1) + require.Equal(t, syncTestCallerNamespace, env.authorizer.targets[0].Namespace) + require.Equal(t, syncTestEndpoint, env.authorizer.targets[0].NexusEndpointName) + require.Empty(t, env.endpointCalls, "a denied caller learns nothing about the endpoint") +} + +func TestExecuteSyncNexusOperation_RefusesATargetThatTakesNoTasks(t *testing.T) { + env := newSyncTestEnv(t, true, true, func(info *persistencespb.NamespaceInfo) { + info.State = enumspb.NAMESPACE_STATE_DELETED + }) + _, err := env.handler.ExecuteSyncNexusOperation(context.Background(), syncTestRequest()) + var invalidState *serviceerror.NamespaceInvalidState + require.ErrorAs(t, err, &invalidState, "the same answer as on the Nexus HTTP route") + require.Empty(t, env.countLimiter.calls) +} + +func TestExecuteSyncNexusOperation_TakesTheTargetConcurrencyLimit(t *testing.T) { + env := newSyncTestEnv(t, true, true) + env.countLimiter.err = serviceerror.NewResourceExhausted( + enumspb.RESOURCE_EXHAUSTED_CAUSE_CONCURRENT_LIMIT, "too many") + _, err := env.handler.ExecuteSyncNexusOperation(context.Background(), syncTestRequest()) + var exhausted *serviceerror.ResourceExhausted + require.ErrorAs(t, err, &exhausted) + require.Equal(t, + []string{syncTestTargetNamespace + " " + configs.DispatchNexusTaskByEndpointAPIName}, + env.countLimiter.calls) + require.Empty(t, env.rateLimiter.calls) +} diff --git a/service/frontend/service.go b/service/frontend/service.go index 3fb7c286339..4c560c72bc0 100644 --- a/service/frontend/service.go +++ b/service/frontend/service.go @@ -213,6 +213,8 @@ type Config struct { MaxConcurrentAdminBatchOperation dynamicconfig.IntPropertyFn EnableBatchOperationsForStandaloneActivities dynamicconfig.BoolPropertyFnWithNamespaceFilter + EnableExecuteSyncNexusOperation dynamicconfig.BoolPropertyFnWithNamespaceFilter + EnableUpdateWorkflowExecution dynamicconfig.BoolPropertyFnWithNamespaceFilter EnableUpdateWorkflowExecutionAsyncAccepted dynamicconfig.BoolPropertyFnWithNamespaceFilter EnableWorkflowUpdateCallbacks dynamicconfig.BoolPropertyFnWithNamespaceFilter @@ -416,6 +418,8 @@ func NewConfig( MaxConcurrentAdminBatchOperation: dynamicconfig.FrontendMaxConcurrentAdminBatchOperation.Get(dc), EnableBatchOperationsForStandaloneActivities: dynamicconfig.FrontendEnableBatchOperationsForStandaloneActivities.Get(dc), + EnableExecuteSyncNexusOperation: dynamicconfig.FrontendEnableExecuteSyncNexusOperation.Get(dc), + EnableUpdateWorkflowExecution: dynamicconfig.FrontendEnableUpdateWorkflowExecution.Get(dc), EnableUpdateWorkflowExecutionAsyncAccepted: dynamicconfig.FrontendEnableUpdateWorkflowExecutionAsyncAccepted.Get(dc), EnableWorkflowUpdateCallbacks: dynamicconfig.EnableWorkflowUpdateCallbacks.Get(dc), diff --git a/service/frontend/workflow_handler.go b/service/frontend/workflow_handler.go index 690a8f7152e..2ef800432db 100644 --- a/service/frontend/workflow_handler.go +++ b/service/frontend/workflow_handler.go @@ -66,6 +66,7 @@ import ( "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/namespace" "go.temporal.io/server/common/namespace/nsreplication" + commonnexus "go.temporal.io/server/common/nexus" "go.temporal.io/server/common/payload" "go.temporal.io/server/common/payloads" "go.temporal.io/server/common/persistence" @@ -169,6 +170,10 @@ type ( httpEnabled bool registry *chasm.Registry workerDeploymentReadRateLimiter quotas.RequestRateLimiter + endpointRegistry commonnexus.EndpointRegistry + namespaceRateLimiter interceptor.NamespaceRateLimitInterceptor + nexusOperationConfig *chasmnexus.Config + syncNexusChecks SyncNexusChecks } ) @@ -350,6 +355,10 @@ func NewWorkflowHandler( registry *chasm.Registry, workerDeploymentReadRateLimiter quotas.RequestRateLimiter, validator *workflow.RequestValidator, + endpointRegistry commonnexus.EndpointRegistry, + namespaceRateLimiter interceptor.NamespaceRateLimitInterceptor, + nexusOperationConfig *chasmnexus.Config, + syncNexusChecks SyncNexusChecks, ) *WorkflowHandler { handler := &WorkflowHandler{ ActivityHandler: activityHandler, @@ -406,6 +415,10 @@ func NewWorkflowHandler( registry: registry, workerDeploymentReadRateLimiter: workerDeploymentReadRateLimiter, validator: validator, + endpointRegistry: endpointRegistry, + namespaceRateLimiter: namespaceRateLimiter, + nexusOperationConfig: nexusOperationConfig, + syncNexusChecks: syncNexusChecks, } return handler diff --git a/service/frontend/workflow_handler_test.go b/service/frontend/workflow_handler_test.go index 2f91d36942e..de7eaa2950b 100644 --- a/service/frontend/workflow_handler_test.go +++ b/service/frontend/workflow_handler_test.go @@ -246,6 +246,10 @@ func (s *WorkflowHandlerSuite) getWorkflowHandler(config *Config) *WorkflowHandl s.mockSearchAttributesMapperProvider, saValidator, ), + nil, // Not testing ExecuteSyncNexusOperation here + nil, + nil, + SyncNexusChecks{}, ) } diff --git a/tests/nexus_sync_test.go b/tests/nexus_sync_test.go new file mode 100644 index 00000000000..763409e3d7a --- /dev/null +++ b/tests/nexus_sync_test.go @@ -0,0 +1,458 @@ +package tests + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/google/uuid" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + failurepb "go.temporal.io/api/failure/v1" + nexuspb "go.temporal.io/api/nexus/v1" + "go.temporal.io/api/operatorservice/v1" + "go.temporal.io/api/serviceerror" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/server/common/api" + "go.temporal.io/server/common/authorization" + "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/namespace" + "go.temporal.io/server/common/testing/parallelsuite" + "go.temporal.io/server/tests/testcore" +) + +const executeSyncNexusOperationAPIName = api.WorkflowServicePrefix + "ExecuteSyncNexusOperation" + +// NexusSyncSuite covers ExecuteSyncNexusOperation, the frontend call that runs one sync Nexus +// operation for a caller namespace and persists nothing. +type NexusSyncSuite struct { + parallelsuite.Suite[*NexusSyncSuite] +} + +func TestNexusSyncSuite(t *testing.T) { + parallelsuite.Run(t, &NexusSyncSuite{}) +} + +// syncNexusTaskAnswer is what a test worker sends back for one Nexus task. Set exactly one field. +type syncNexusTaskAnswer struct { + response *nexuspb.Response + failure *failurepb.Failure +} + +func (s *NexusSyncSuite) newEnv(opts ...testcore.TestOption) *testcore.TestEnv { + opts = append([]testcore.TestOption{ + testcore.WithDynamicConfig(dynamicconfig.FrontendEnableExecuteSyncNexusOperation, true), + }, opts...) + return testcore.NewEnv(s.T(), opts...) +} + +// registerNamespace registers another namespace, so the caller and the handler namespaces differ +// as they do for a stream reader. +func (s *NexusSyncSuite) registerNamespace(env *testcore.TestEnv, prefix string) namespace.Name { + name := namespace.Name(testcore.RandomizeStr(prefix)) + _, err := env.RegisterNamespace(s.Context(), name, 1, enumspb.ARCHIVAL_STATE_DISABLED, "", "") + s.NoError(err) + return name +} + +// registerRateLimitedNamespace registers a namespace with two requests per second and no burst +// above that. The limit is set first, since the frontend reads it when it first sees the +// namespace. Registration itself makes a few calls, so the limit can't be lower. +func (s *NexusSyncSuite) registerRateLimitedNamespace(env *testcore.TestEnv) namespace.Name { + name := namespace.Name(testcore.RandomizeStr("sync-limited")) + constraint := dynamicconfig.Constraints{Namespace: name.String()} + s.T().Cleanup(env.OverrideDynamicConfig(dynamicconfig.FrontendMaxNamespaceRPSPerInstance, + []dynamicconfig.ConstrainedValue{{Constraints: constraint, Value: 2}})) + s.T().Cleanup(env.OverrideDynamicConfig(dynamicconfig.FrontendMaxNamespaceBurstRatioPerInstance, + []dynamicconfig.ConstrainedValue{{Constraints: constraint, Value: 1.0}})) + _, err := env.RegisterNamespace(s.Context(), name, 1, enumspb.ARCHIVAL_STATE_DISABLED, "", "") + s.NoError(err) + return name +} + +func (s *NexusSyncSuite) createEndpoint( + env *testcore.TestEnv, + targetNamespace namespace.Name, + taskQueue string, +) string { + name := testcore.RandomizedNexusEndpoint(s.T().Name()) + resp, err := env.OperatorClient(). + CreateNexusEndpoint(s.Context(), &operatorservice.CreateNexusEndpointRequest{ + Spec: &nexuspb.EndpointSpec{ + Name: name, + Target: &nexuspb.EndpointTarget{ + Variant: &nexuspb.EndpointTarget_Worker_{ + Worker: &nexuspb.EndpointTarget_Worker{ + Namespace: targetNamespace.String(), + TaskQueue: taskQueue, + }, + }, + }, + }, + }) + s.NoError(err) + s.T().Cleanup(func() { + _, _ = env.OperatorClient(). + DeleteNexusEndpoint(testcore.NewContext(), &operatorservice.DeleteNexusEndpointRequest{ + Id: resp.Endpoint.Id, + Version: resp.Endpoint.Version, + }) + }) + return name +} + +// answerOneTask polls one Nexus task in the target namespace and answers it. The task goes to +// seen before the answer, so the test can check what the handler got. +func (s *NexusSyncSuite) answerOneTask( + env *testcore.TestEnv, + targetNamespace namespace.Name, + taskQueue string, + answer syncNexusTaskAnswer, + seen chan<- *workflowservice.PollNexusTaskQueueResponse, +) <-chan error { + ctx := s.Context() + errCh := make(chan error, 1) + go func() { + res, err := env.FrontendClient(). + PollNexusTaskQueue(ctx, &workflowservice.PollNexusTaskQueueRequest{ + Namespace: targetNamespace.String(), + Identity: uuid.NewString(), + TaskQueue: &taskqueuepb.TaskQueue{ + Name: taskQueue, + Kind: enumspb.TASK_QUEUE_KIND_NORMAL, + }, + }) + if err != nil { + errCh <- err + return + } + if res.GetTaskToken() == nil { + errCh <- errors.New("poll returned no task") + return + } + seen <- res + if answer.failure != nil { + _, err = env.FrontendClient(). + RespondNexusTaskFailed(ctx, &workflowservice.RespondNexusTaskFailedRequest{ + Namespace: targetNamespace.String(), + Identity: uuid.NewString(), + TaskToken: res.GetTaskToken(), + Failure: answer.failure, + }) + } else { + req := &workflowservice.RespondNexusTaskCompletedRequest{ + Namespace: targetNamespace.String(), + Identity: uuid.NewString(), + TaskToken: res.GetTaskToken(), + Response: answer.response, + } + _, err = env.FrontendClient().RespondNexusTaskCompleted(ctx, req) + } + errCh <- err + }() + return errCh +} + +func syncStartAnswer(start *nexuspb.StartOperationResponse) syncNexusTaskAnswer { + return syncNexusTaskAnswer{response: &nexuspb.Response{ + Variant: &nexuspb.Response_StartOperation{StartOperation: start}, + }} +} + +func syncSuccessAnswer(data string) syncNexusTaskAnswer { + return syncStartAnswer(&nexuspb.StartOperationResponse{ + Variant: &nexuspb.StartOperationResponse_SyncSuccess{ + SyncSuccess: &nexuspb.StartOperationResponse_Sync{ + Payload: &commonpb.Payload{Data: []byte(data)}, + }, + }, + }) +} + +func syncNexusRequest( + callerNamespace namespace.Name, + endpoint string, +) *workflowservice.ExecuteSyncNexusOperationRequest { + return &workflowservice.ExecuteSyncNexusOperationRequest{ + Namespace: callerNamespace.String(), + Identity: "sync-test", + RequestId: uuid.NewString(), + Endpoint: endpoint, + Service: "temporal.sdk.streams.v1.TemporalStreams", + Operation: "read", + Input: &commonpb.Payload{Data: []byte(`{"stream":"s1"}`)}, + NexusHeader: map[string]string{"X-Trace-Id": "trace-1"}, + } +} + +func (s *NexusSyncSuite) requireOneResourceExhausted(errs []error) { + for _, err := range errs { + if _, ok := errors.AsType[*serviceerror.ResourceExhausted](err); ok { + return + } + } + s.Failf("no call was rate limited", "errors: %v", errs) +} + +func (s *NexusSyncSuite) TestSyncResult() { + env := s.newEnv() + target := s.registerNamespace(env, "sync-target") + taskQueue := testcore.RandomizeStr("sync-tq") + endpoint := s.createEndpoint(env, target, taskQueue) + + seen := make(chan *workflowservice.PollNexusTaskQueueResponse, 1) + done := s.answerOneTask(env, target, taskQueue, syncSuccessAnswer(`{"records":[]}`), seen) + + req := syncNexusRequest(env.Namespace(), endpoint) + resp, err := env.FrontendClient().ExecuteSyncNexusOperation(s.Context(), req) + s.NoError(err) + s.JSONEq(`{"records":[]}`, string(resp.GetResult().GetData())) + s.NoError(s.Rcv(done)) + + task := s.Rcv(seen) + s.Equal(endpoint, task.GetRequest().GetEndpoint()) + s.Equal("trace-1", task.GetRequest().GetHeader()["x-trace-id"]) + start := task.GetRequest().GetStartOperation() + s.Equal("temporal.sdk.streams.v1.TemporalStreams", start.GetService()) + s.Equal("read", start.GetOperation()) + s.Equal(req.GetRequestId(), start.GetRequestId()) + s.Empty(start.GetCallback()) + s.JSONEq(`{"stream":"s1"}`, string(start.GetPayload().GetData())) +} + +func (s *NexusSyncSuite) TestHandlerError() { + env := s.newEnv() + taskQueue := testcore.RandomizeStr("sync-tq") + endpoint := s.createEndpoint(env, env.Namespace(), taskQueue) + + seen := make(chan *workflowservice.PollNexusTaskQueueResponse, 1) + done := s.answerOneTask( + env, + env.Namespace(), + taskQueue, + syncNexusTaskAnswer{failure: &failurepb.Failure{ + Message: "stream name is not valid", + FailureInfo: &failurepb.Failure_NexusHandlerFailureInfo{ + NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{ + Type: "BAD_REQUEST", + RetryBehavior: enumspb.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE, + }, + }, + }}, + seen, + ) + + resp, err := env.FrontendClient(). + ExecuteSyncNexusOperation(s.Context(), syncNexusRequest(env.Namespace(), endpoint)) + s.NoError(err) + s.NoError(s.Rcv(done)) + s.Nil(resp.GetResult()) + s.Equal("stream name is not valid", resp.GetFailure().GetMessage()) + s.Equal("BAD_REQUEST", resp.GetFailure().GetNexusHandlerFailureInfo().GetType()) +} + +func (s *NexusSyncSuite) TestOperationFailure() { + env := s.newEnv() + taskQueue := testcore.RandomizeStr("sync-tq") + endpoint := s.createEndpoint(env, env.Namespace(), taskQueue) + + seen := make(chan *workflowservice.PollNexusTaskQueueResponse, 1) + done := s.answerOneTask( + env, + env.Namespace(), + taskQueue, + syncStartAnswer(&nexuspb.StartOperationResponse{ + Variant: &nexuspb.StartOperationResponse_Failure{ + Failure: &failurepb.Failure{ + Message: "replay range lost", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{ + Type: "StreamReplayRangeLostError", + NonRetryable: true, + }, + }, + }, + }, + }), + seen, + ) + + resp, err := env.FrontendClient(). + ExecuteSyncNexusOperation(s.Context(), syncNexusRequest(env.Namespace(), endpoint)) + s.NoError(err) + s.NoError(s.Rcv(done)) + s.Equal("replay range lost", resp.GetFailure().GetMessage()) + s.Equal("StreamReplayRangeLostError", resp.GetFailure().GetApplicationFailureInfo().GetType()) +} + +func (s *NexusSyncSuite) TestAsyncAnswerRefused() { + env := s.newEnv() + taskQueue := testcore.RandomizeStr("sync-tq") + endpoint := s.createEndpoint(env, env.Namespace(), taskQueue) + + seen := make(chan *workflowservice.PollNexusTaskQueueResponse, 1) + done := s.answerOneTask( + env, + env.Namespace(), + taskQueue, + syncStartAnswer(&nexuspb.StartOperationResponse{ + Variant: &nexuspb.StartOperationResponse_AsyncSuccess{ + AsyncSuccess: &nexuspb.StartOperationResponse_Async{OperationToken: "token"}, + }, + }), + seen, + ) + + _, err := env.FrontendClient(). + ExecuteSyncNexusOperation(s.Context(), syncNexusRequest(env.Namespace(), endpoint)) + s.NoError(s.Rcv(done)) + var precondition *serviceerror.FailedPrecondition + s.ErrorAs(err, &precondition) +} + +func (s *NexusSyncSuite) TestUnknownEndpoint() { + env := s.newEnv() + req := syncNexusRequest(env.Namespace(), "no-such-endpoint") + _, err := env.FrontendClient().ExecuteSyncNexusOperation(s.Context(), req) + var notFound *serviceerror.NotFound + s.ErrorAs(err, ¬Found) +} + +func (s *NexusSyncSuite) TestNoPollerDeadline() { + env := s.newEnv() + endpoint := s.createEndpoint(env, env.Namespace(), testcore.RandomizeStr("sync-tq")) + + ctx, cancel := context.WithTimeout(s.Context(), 2*time.Second) + defer cancel() + _, err := env.FrontendClient(). + ExecuteSyncNexusOperation(ctx, syncNexusRequest(env.Namespace(), endpoint)) + var deadline *serviceerror.DeadlineExceeded + s.ErrorAs(err, &deadline) +} + +func (s *NexusSyncSuite) TestDisabled() { + env := testcore.NewEnv(s.T()) + endpoint := s.createEndpoint(env, env.Namespace(), testcore.RandomizeStr("sync-tq")) + _, err := env.FrontendClient(). + ExecuteSyncNexusOperation(s.Context(), syncNexusRequest(env.Namespace(), endpoint)) + var unimplemented *serviceerror.Unimplemented + s.ErrorAs(err, &unimplemented) +} + +// TestAuthorizationAndRateLimits needs a dedicated cluster for the authorizer hook. The rate +// limit cases share it, since they set limits on namespaces other than the test namespace. +func (s *NexusSyncSuite) TestAuthorizationAndRateLimits() { + env := s.newEnv(testcore.WithDedicatedCluster()) + target := s.registerNamespace(env, "sync-target") + + s.RunSequential("AuthorizesTheCallerNamespace", func(s *NexusSyncSuite) { + taskQueue := testcore.RandomizeStr("sync-tq") + endpoint := s.createEndpoint(env, target, taskQueue) + + targets := make(chan *authorization.CallTarget, 10) + env.SetOnAuthorize( + func( + _ context.Context, + _ *authorization.Claims, + ct *authorization.CallTarget, + ) (authorization.Result, error) { + if ct.APIName != executeSyncNexusOperationAPIName { + return authorization.Result{Decision: authorization.DecisionAllow}, nil + } + targets <- ct + // Deny the target namespace. The call must still pass, since the caller is checked. + if ct.Namespace == target.String() { + return authorization.Result{Decision: authorization.DecisionDeny}, nil + } + return authorization.Result{Decision: authorization.DecisionAllow}, nil + }, + ) + defer env.SetOnAuthorize(nil) + + seen := make(chan *workflowservice.PollNexusTaskQueueResponse, 1) + done := s.answerOneTask(env, target, taskQueue, syncSuccessAnswer(`{}`), seen) + _, err := env.FrontendClient(). + ExecuteSyncNexusOperation(s.Context(), syncNexusRequest(env.Namespace(), endpoint)) + s.NoError(err) + s.NoError(s.Rcv(done)) + s.Equal(env.Namespace().String(), s.Rcv(targets).Namespace) + }) + + s.RunSequential("DeniedCallerNamespace", func(s *NexusSyncSuite) { + endpoint := s.createEndpoint(env, target, testcore.RandomizeStr("sync-tq")) + env.SetOnAuthorize( + func( + _ context.Context, + _ *authorization.Claims, + ct *authorization.CallTarget, + ) (authorization.Result, error) { + if ct.APIName == executeSyncNexusOperationAPIName && + ct.Namespace == env.Namespace().String() { + return authorization.Result{Decision: authorization.DecisionDeny}, nil + } + return authorization.Result{Decision: authorization.DecisionAllow}, nil + }, + ) + defer env.SetOnAuthorize(nil) + + _, err := env.FrontendClient(). + ExecuteSyncNexusOperation(s.Context(), syncNexusRequest(env.Namespace(), endpoint)) + var denied *serviceerror.PermissionDenied + s.ErrorAs(err, &denied) + }) + + s.RunSequential("AuthorizesTheEndpointName", func(s *NexusSyncSuite) { + endpoint := s.createEndpoint(env, target, testcore.RandomizeStr("sync-tq")) + env.SetOnAuthorize( + func( + _ context.Context, + _ *authorization.Claims, + ct *authorization.CallTarget, + ) (authorization.Result, error) { + // An authorizer that limits endpoints by name, as on the Nexus HTTP route. + if ct.NexusEndpointName == endpoint { + return authorization.Result{Decision: authorization.DecisionDeny}, nil + } + return authorization.Result{Decision: authorization.DecisionAllow}, nil + }, + ) + defer env.SetOnAuthorize(nil) + + _, err := env.FrontendClient(). + ExecuteSyncNexusOperation(s.Context(), syncNexusRequest(env.Namespace(), endpoint)) + var denied *serviceerror.PermissionDenied + s.ErrorAs(err, &denied) + }) + + s.RunSequential("TargetNamespaceRateLimitCounts", func(s *NexusSyncSuite) { + limited := s.registerRateLimitedNamespace(env) + endpoint := s.createEndpoint(env, limited, testcore.RandomizeStr("sync-tq")) + + // Nobody polls, so each allowed call waits for its short deadline. The calls take tokens + // faster than the bucket fills, so one of them finds it empty. + var errs []error + for range 5 { + ctx, cancel := context.WithTimeout(s.Context(), 200*time.Millisecond) + _, err := env.FrontendClient(). + ExecuteSyncNexusOperation(ctx, syncNexusRequest(env.Namespace(), endpoint)) + cancel() + errs = append(errs, err) + } + s.requireOneResourceExhausted(errs) + }) + + s.RunSequential("CallerNamespaceRateLimitCounts", func(s *NexusSyncSuite) { + limited := s.registerRateLimitedNamespace(env) + + // An unknown endpoint answers at once, so the calls come faster than two per second. + var errs []error + for range 5 { + req := syncNexusRequest(limited, "no-such-endpoint") + _, err := env.FrontendClient().ExecuteSyncNexusOperation(s.Context(), req) + errs = append(errs, err) + } + s.requireOneResourceExhausted(errs) + }) +}