Fix mislabeled "system payload" issue

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
 }

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,

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")
+}

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{