diff --git a/chasm/lib/tests/nexus_service.go b/chasm/lib/tests/nexus_service.go index 0ae0d42905f..386073e1ec6 100644 --- a/chasm/lib/tests/nexus_service.go +++ b/chasm/lib/tests/nexus_service.go @@ -9,9 +9,14 @@ import ( "go.temporal.io/server/common/payload" ) -var TestOperation = nexus.NewSyncOperation("TestOperation", func(ctx context.Context, input string, options nexus.StartOperationOptions) (string, error) { - return "Hello, " + input, nil -}) +// TestOperation returns a failurepb.Failure so that it is a proper protobuf type. Its used for +// testing the System Nexus Endpoint, which only accepts protobuf-encoded payloads. +var TestOperation = nexus.NewSyncOperation( + "TestOperation", + func(ctx context.Context, input string, options nexus.StartOperationOptions) (*commonpb.DataBlob, error) { + d := []byte("Hello, " + input) + return &commonpb.DataBlob{Data: d}, nil + }) // TestOperationWithPayload is identical to TestOperation, except its response embeds a // nested *commonpb.Payload. It exists to exercise the commonnexus.SystemPayloadMetadataKey @@ -20,10 +25,18 @@ var TestOperationWithPayload = nexus.NewSyncOperation("TestOperationWithPayload" return &commonpb.Payloads{Payloads: []*commonpb.Payload{payload.EncodeString("Hello, " + input)}}, nil }) +// TestOperationStringOutput returns a string, which the data converter encodes as JSON rather +// than protobuf. It exists to exercise the System Nexus Endpoint's rejection of non-protobuf +// responses in service/history/handler.go's StartNexusOperation. +var TestOperationStringOutput = nexus.NewSyncOperation("TestOperationStringOutput", func(ctx context.Context, input string, options nexus.StartOperationOptions) (string, error) { + return "Hello, " + input, nil +}) + func NewTestServiceNexusService() *nexus.Service { service := nexus.NewService("TestService") service.MustRegister(TestOperation) service.MustRegister(TestOperationWithPayload) + service.MustRegister(TestOperationStringOutput) return service } @@ -43,5 +56,6 @@ func NewTestServiceNexusServiceProcessor() *chasm.NexusServiceProcessor { sp := chasm.NewNexusServiceProcessor("TestService") sp.MustRegisterOperation("TestOperation", chasm.NewRegisterableNexusOperationProcessor(testOperationProcessor{})) sp.MustRegisterOperation("TestOperationWithPayload", chasm.NewRegisterableNexusOperationProcessor(testOperationProcessor{})) + sp.MustRegisterOperation("TestOperationStringOutput", chasm.NewRegisterableNexusOperationProcessor(testOperationProcessor{})) return sp } diff --git a/service/history/handler.go b/service/history/handler.go index 42e88cfc4d5..64cba0eba80 100644 --- a/service/history/handler.go +++ b/service/history/handler.go @@ -2540,6 +2540,9 @@ func (h *Handler) UnpauseWorkflowExecution(ctx context.Context, request *history return unpauseResp, nil } +// StartNexusOperation is the History Service's StartNexusOperation endpoint is for dispatching requests +// to the System Nexus Endpoint, distinct from the Frontend Service's `nexus_handler.go` which +// starts Nexus operations by sending Nexus tasks directly to workers via the Matching Service. func (h *Handler) StartNexusOperation( ctx context.Context, req *historyservice.StartNexusOperationRequest, @@ -2602,16 +2605,29 @@ func (h *Handler) StartNexusOperation( h.logger.Error("failed to encode payload", tag.Error(err), tag.RequestID(requestID)) return nil, serviceerror.NewInternal("internal error (request ID: " + requestID + ")") } + var payload *commonpb.Payload if len(ps.GetPayloads()) == 1 { payload = ps.GetPayloads()[0] } + // Responses from the System Nexus Endpoint are server generated, so we must mark them as system payloads. if payload != nil { if payload.Metadata == nil { payload.Metadata = make(map[string][]byte, 1) } + + // For now, we require all responess from the System Nexus Endpoint be protobufs. + encoding := string(payload.Metadata["encoding"]) + if encoding != "binary/protobuf" { + return nil, serviceerror.NewFailedPreconditionf("system payload must be encoded as binary/protobuf but got %s", encoding) + } + if _, ok := payload.Metadata["messageType"]; !ok { + return nil, serviceerror.NewFailedPrecondition("system payload missing messageType metadata key") + } + payload.Metadata[commonnexus.SystemPayloadMetadataKey] = []byte("true") } + response.Variant = &nexuspb.StartOperationResponse_SyncSuccess{ SyncSuccess: &nexuspb.StartOperationResponse_Sync{ Payload: payload, diff --git a/service/history/handler_test.go b/service/history/handler_test.go index e0c3df1b9f0..3394f59206e 100644 --- a/service/history/handler_test.go +++ b/service/history/handler_test.go @@ -450,3 +450,31 @@ func TestStartNexusOperation_SystemNexusEndpointPayloadMetadataFlag(t *testing.T }) } } + +func TestStartNexusOperation_SystemNexusEndpointRejectsNonProtoResponse(t *testing.T) { + registry := nexus.NewServiceRegistry() + registry.MustRegister(chasmtests.NewTestServiceNexusService()) + nexusHandler, err := registry.NewHandler() + require.NoError(t, err) + + h := Handler{ + logger: log.NewNoopLogger(), + nexusHandler: nexusHandler, + } + + // TestOperationStringOutput returns a string, which the data converter encodes as JSON + // instead of protobuf. The System Nexus Endpoint only accepts protobuf-encoded responses. + resp, err := h.StartNexusOperation(context.Background(), &historyservice.StartNexusOperationRequest{ + Request: &nexuspb.StartOperationRequest{ + Service: "TestService", + Operation: "TestOperationStringOutput", + RequestId: "test-request-id", + Payload: payload.EncodeString("Temporal"), + }, + }) + require.Nil(t, resp) + + var failedPrecondition *serviceerror.FailedPrecondition + require.ErrorAs(t, err, &failedPrecondition) + require.ErrorContains(t, err, "system payload must be encoded as binary/protobuf but got json/plain") +} diff --git a/tests/nexus_workflow_test.go b/tests/nexus_workflow_test.go index 699395d9735..abef91f61e1 100644 --- a/tests/nexus_workflow_test.go +++ b/tests/nexus_workflow_test.go @@ -55,6 +55,12 @@ import ( "google.golang.org/protobuf/types/known/durationpb" ) +const ( + dataBlobMessageType = "temporal.api.common.v1.DataBlob" + payloadsMessageType = "temporal.api.common.v1.Payloads" + protobufEncoding = "binary/protobuf" +) + type NexusWorkflowTestSuite struct { parallelsuite.Suite[*NexusWorkflowTestSuite] } @@ -3314,6 +3320,9 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationSystemEndpoint(chasmEnabled b result := completedEvent.GetNexusOperationCompletedEventAttributes().Result s.NotNil(result) s.Equal([]byte("true"), result.GetMetadata()[commonnexus.SystemPayloadMetadataKey]) + // TestOperation returns a proto message, so the result must be proto encoded, not JSON. + s.Equal([]byte(dataBlobMessageType), result.GetMetadata()["messageType"]) + s.Equal([]byte(protobufEncoding), result.GetMetadata()["encoding"]) // Complete the workflow _, err = env.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{ @@ -3333,9 +3342,10 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationSystemEndpoint(chasmEnabled b }, }) s.NoError(err) - var response string + var response commonpb.DataBlob s.NoError(run.Get(ctx, &response)) - s.Equal("Hello, Temporal", response) + data := response.Data + s.Equal("Hello, Temporal", string(data)) } // NOTE: This test cannot use the SDK workflow package because there is a restriction that prevents setting the @@ -3392,9 +3402,10 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationSystemEndpoint_PayloadMetadat completedEvent := s.RequireHistoryEvent(pollResp.History.Events, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED) result := completedEvent.GetNexusOperationCompletedEventAttributes().Result s.NotNil(result) - // TestOperationWithPayload's response embeds a nested Payload, so the system payload metadata - // flag must be set. + // TestOperationWithPayload's response embeds a nested Payload, so the system payload metadata flag must be set. s.Equal([]byte("true"), result.GetMetadata()[commonnexus.SystemPayloadMetadataKey]) + s.Equal([]byte(payloadsMessageType), result.GetMetadata()["messageType"]) + s.Equal([]byte(protobufEncoding), result.GetMetadata()["encoding"]) // Complete the workflow _, err = env.FrontendClient().RespondWorkflowTaskCompleted(s.Context(), &workflowservice.RespondWorkflowTaskCompletedRequest{