diff --git a/cmd/proxygenerator/interceptor.go b/cmd/proxygenerator/interceptor.go index 1a6f8830..c9d046b3 100644 --- a/cmd/proxygenerator/interceptor.go +++ b/cmd/proxygenerator/interceptor.go @@ -127,7 +127,7 @@ var failureTypes = []string{ {{ range $i, $name := .GrpcFailure }}{{ if $i }}, { // NewPayloadVisitorInterceptor creates a new gRPC interceptor for workflowservice messages. // -// Note: Failure converters should come before payload codec converts, to allow the +// Note: Failure converters should come before payload codec converts, to allow the // payloads generated by the failure convert to be intercepted by the payload codec converters. func NewPayloadVisitorInterceptor(options PayloadVisitorInterceptorOptions) (grpc.UnaryClientInterceptor, error) { return func(ctx context.Context, method string, req, response interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { @@ -135,7 +135,7 @@ func NewPayloadVisitorInterceptor(options PayloadVisitorInterceptorOptions) (grp err := VisitPayloads(ctx, reqMsg, *options.Outbound) if err != nil { return err - } + } } err := invoker(ctx, method, req, response, cc, opts...) @@ -153,7 +153,7 @@ func NewPayloadVisitorInterceptor(options PayloadVisitorInterceptorOptions) (grp err = visitErr } } - + return err }, nil } @@ -204,7 +204,7 @@ type FailureVisitorInterceptorOptions struct { // NewFailureVisitorInterceptor creates a new gRPC interceptor for workflowservice messages. // -// Note: Failure converters should come before payload codec converts, to allow the +// Note: Failure converters should come before payload codec converts, to allow the // payloads generated by the failure convert to be intercepted by the payload codec converters. func NewFailureVisitorInterceptor(options FailureVisitorInterceptorOptions) (grpc.UnaryClientInterceptor, error) { return func(ctx context.Context, method string, req, response interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { @@ -212,7 +212,7 @@ func NewFailureVisitorInterceptor(options FailureVisitorInterceptorOptions) (grp err := VisitFailures(ctx, reqMsg, *options.Outbound) if err != nil { return err - } + } } err := invoker(ctx, method, req, response, cc, opts...) @@ -230,7 +230,7 @@ func NewFailureVisitorInterceptor(options FailureVisitorInterceptorOptions) (grp err = visitErr } } - + return err }, nil } @@ -297,6 +297,9 @@ func visitPayload( concState *payloadConcurrencyState, fieldPtr **common.Payload, ) error { + if handled, err := visitSystemPayload(ctx, options, concState, *fieldPtr); handled { + return err + } if concState != nil { if errPtr := concState.firstErr.Load(); errPtr != nil { return *errPtr @@ -538,7 +541,7 @@ func visitPayloads( if options.SkipSearchAttributes { continue } {{end}} if o == nil { continue } - + prevCtx := ctx.Context if options.ContextHook != nil { var hookErr error @@ -559,14 +562,6 @@ func visitPayloads( if err := visitPayload(ctx, options, o, concState, &result); err != nil { return err } o.Outcome = &workflowservice.PollNexusOperationExecutionResponse_Result{Result: result} } - {{else if and (eq $type "*command.ScheduleNexusOperationCommandAttributes") (eq . "Input")}} - if o.Input != nil { - if o.GetEndpoint() == "__temporal_system" { - if err := visitSystemNexusEnvelope(ctx, options, concState, o); err != nil { return err } - } else { - if err := visitPayload(ctx, options, o, concState, &o.Input); err != nil { return err } - } - } {{else}} if o.{{.}} != nil { if err := visitPayload(ctx, options, o, concState, &o.{{.}}); err != nil { return err } diff --git a/proxy/interceptor.go b/proxy/interceptor.go index edad35ec..682a3dec 100644 --- a/proxy/interceptor.go +++ b/proxy/interceptor.go @@ -293,6 +293,9 @@ func visitPayload( concState *payloadConcurrencyState, fieldPtr **common.Payload, ) error { + if handled, err := visitSystemPayload(ctx, options, concState, *fieldPtr); handled { + return err + } if concState != nil { if errPtr := concState.firstErr.Load(); errPtr != nil { return *errPtr @@ -998,14 +1001,8 @@ func visitPayloads( } if o.Input != nil { - if o.GetEndpoint() == "__temporal_system" { - if err := visitSystemNexusEnvelope(ctx, options, concState, o); err != nil { - return err - } - } else { - if err := visitPayload(ctx, options, o, concState, &o.Input); err != nil { - return err - } + if err := visitPayload(ctx, options, o, concState, &o.Input); err != nil { + return err } } diff --git a/proxy/interceptor_test.go b/proxy/interceptor_test.go index c3d314a1..9dedd8a0 100644 --- a/proxy/interceptor_test.go +++ b/proxy/interceptor_test.go @@ -423,6 +423,51 @@ func TestClientInterceptor(t *testing.T) { require.True(proto.Equal(inputs.Payloads[0], inboundPayload)) } +func TestClientInterceptorVisitsNestedSystemPayloads(t *testing.T) { + server, err := startTestGRPCServer() + require.NoError(t, err) + defer server.Stop() + + var outboundVisits, inboundVisits int + interceptor, err := NewPayloadVisitorInterceptor(PayloadVisitorInterceptorOptions{ + Outbound: &VisitPayloadsOptions{Visitor: func(ctx *VisitPayloadsContext, payloads []*common.Payload) ([]*common.Payload, error) { + outboundVisits += len(payloads) + return collectVisitor(ctx, payloads) + }}, + Inbound: &VisitPayloadsOptions{Visitor: func(ctx *VisitPayloadsContext, payloads []*common.Payload) ([]*common.Payload, error) { + inboundVisits += len(payloads) + return collectVisitor(ctx, payloads) + }}, + }) + require.NoError(t, err) + + conn, err := grpc.Dial(server.addr, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithChainUnaryInterceptor(interceptor), + ) + require.NoError(t, err) + defer conn.Close() + client := workflowservice.NewWorkflowServiceClient(conn) + + scheduledCommand := buildSystemPayloadCommand(t, binaryProtobufEncoding, signalWithStartType) + _, err = client.RespondWorkflowTaskCompleted(context.Background(), &workflowservice.RespondWorkflowTaskCompletedRequest{ + Commands: []*command.Command{scheduledCommand}, + }) + require.NoError(t, err) + require.NotNil(t, server.respondWorkflowTaskCompletedRequest) + outer := server.respondWorkflowTaskCompletedRequest.Commands[0].GetScheduleNexusOperationCommandAttributes().Input + require.True(t, proto.Equal(visitedSignalWithStartRequest, decodeEnvelope(t, outer))) + + response, err := client.PollWorkflowTaskQueue(context.Background(), &workflowservice.PollWorkflowTaskQueueRequest{}) + require.NoError(t, err) + inboundOuter := response.History.Events[0].GetNexusOperationCompletedEventAttributes().Result + require.True(t, proto.Equal(visitedSignalWithStartRequest, decodeEnvelope(t, inboundOuter))) + // SignalWithStart has seven nested payloads. The outer protobuf envelopes + // are never passed to either codec visitor. + require.Equal(t, 7, outboundVisits) + require.Equal(t, 7, inboundVisits) +} + func TestClientInterceptorGrpcFailures(t *testing.T) { require := require.New(t) @@ -507,10 +552,11 @@ func TestClientInterceptorGrpcFailures(t *testing.T) { type testGRPCServer struct { workflowservice.UnimplementedWorkflowServiceServer *grpc.Server - listener net.Listener - addr string - startWorkflowExecutionRequest *workflowservice.StartWorkflowExecutionRequest - startWorkflowExecutionMetadata metadata.MD + listener net.Listener + addr string + startWorkflowExecutionRequest *workflowservice.StartWorkflowExecutionRequest + startWorkflowExecutionMetadata metadata.MD + respondWorkflowTaskCompletedRequest *workflowservice.RespondWorkflowTaskCompletedRequest } func startTestGRPCServer() (*testGRPCServer, error) { @@ -583,6 +629,36 @@ func (t *testGRPCServer) PollActivityTaskQueue( }, nil } +func (t *testGRPCServer) RespondWorkflowTaskCompleted( + ctx context.Context, + req *workflowservice.RespondWorkflowTaskCompletedRequest, +) (*workflowservice.RespondWorkflowTaskCompletedResponse, error) { + t.respondWorkflowTaskCompletedRequest = req + return &workflowservice.RespondWorkflowTaskCompletedResponse{}, nil +} + +func (t *testGRPCServer) PollWorkflowTaskQueue( + ctx context.Context, + req *workflowservice.PollWorkflowTaskQueueRequest, +) (*workflowservice.PollWorkflowTaskQueueResponse, error) { + data, err := proto.Marshal(signalWithStartRequest) + if err != nil { + return nil, err + } + return &workflowservice.PollWorkflowTaskQueueResponse{History: &history.History{Events: []*history.HistoryEvent{{ + Attributes: &history.HistoryEvent_NexusOperationCompletedEventAttributes{ + NexusOperationCompletedEventAttributes: &history.NexusOperationCompletedEventAttributes{Result: &common.Payload{ + Data: data, + Metadata: map[string][]byte{ + SystemPayloadMetadataKey: []byte(systemPayloadMarkerValue), + "encoding": []byte(binaryProtobufEncoding), + "messageType": []byte(signalWithStartType), + }, + }}, + }, + }}}}, nil +} + func (t *testGRPCServer) ExecuteMultiOperation( ctx context.Context, req *workflowservice.ExecuteMultiOperationRequest) (*workflowservice.ExecuteMultiOperationResponse, error) { diff --git a/proxy/system_nexus.go b/proxy/system_nexus.go deleted file mode 100644 index 81350e53..00000000 --- a/proxy/system_nexus.go +++ /dev/null @@ -1,72 +0,0 @@ -package proxy - -import ( - "fmt" - - "go.temporal.io/api/command/v1" - "google.golang.org/protobuf/proto" - "google.golang.org/protobuf/reflect/protoreflect" - "google.golang.org/protobuf/reflect/protoregistry" -) - -// visitSystemNexusEnvelope decodes the system Nexus envelope in attrs.Input, -// visits the payloads inside the decoded request message, and re-encodes it. -// -// The envelope's proto message type is taken from the payload's "messageType" -// metadata, so no operation registry is required. The envelope must be encoded -// as binary/protobuf. The inner payloads (and only those) are passed to the -// visitor, so external storage and codecs apply to them and not to the envelope -// itself, which is never offloaded or codec-encoded. -func visitSystemNexusEnvelope( - ctx *VisitPayloadsContext, - options *VisitPayloadsOptions, - concState *payloadConcurrencyState, - attrs *command.ScheduleNexusOperationCommandAttributes, -) error { - input := attrs.Input - - if encoding := string(input.GetMetadata()["encoding"]); encoding != "binary/protobuf" { - return fmt.Errorf( - "system nexus envelope for operation %q must be encoded as binary/protobuf but got %q", - attrs.GetOperation(), encoding, - ) - } - - messageType := string(input.GetMetadata()["messageType"]) - if messageType == "" { - return fmt.Errorf( - "system nexus envelope for operation %q is missing the messageType metadata", - attrs.GetOperation(), - ) - } - - mt, err := protoregistry.GlobalTypes.FindMessageByName(protoreflect.FullName(messageType)) - if err != nil { - return fmt.Errorf( - "system nexus envelope for operation %q references unknown message type %q: %w", - attrs.GetOperation(), messageType, err, - ) - } - msg := mt.New().Interface() - - if err := proto.Unmarshal(input.GetData(), msg); err != nil { - return fmt.Errorf( - "failed to unmarshal system nexus envelope for operation %q: %w", - attrs.GetOperation(), err, - ) - } - - if err := visitPayloadsAndWait(ctx, options, msg, concState, msg); err != nil { - return err - } - - data, err := proto.Marshal(msg) - if err != nil { - return fmt.Errorf( - "failed to marshal system nexus envelope for operation %q: %w", - attrs.GetOperation(), err, - ) - } - input.Data = data - return nil -} diff --git a/proxy/system_nexus_test.go b/proxy/system_nexus_test.go index a930a9b9..6c9e60a0 100644 --- a/proxy/system_nexus_test.go +++ b/proxy/system_nexus_test.go @@ -7,6 +7,7 @@ import ( "github.com/stretchr/testify/require" command "go.temporal.io/api/command/v1" common "go.temporal.io/api/common/v1" + history "go.temporal.io/api/history/v1" sdk "go.temporal.io/api/sdk/v1" workflowservice "go.temporal.io/api/workflowservice/v1" "google.golang.org/protobuf/proto" @@ -64,23 +65,24 @@ var visitedSignalWithStartRequest = &workflowservice.SignalWithStartWorkflowExec }, } -// buildSystemNexusCommand builds a system Nexus command whose Input is +// buildSystemPayloadCommand builds a command whose Input is // a SignalWithStartWorkflowExecutionRequest. // The encoding and messageType parameters represent the "encoding" and // "messageType" metadata of the Payload. -func buildSystemNexusCommand(t *testing.T, encoding, messageType string) *command.Command { +func buildSystemPayloadCommand(t *testing.T, encoding, messageType string) *command.Command { t.Helper() data, err := proto.Marshal(signalWithStartRequest) require.NoError(t, err) input := &common.Payload{ Data: data, Metadata: map[string][]byte{ - "encoding": []byte(encoding), - "messageType": []byte(messageType), + SystemPayloadMetadataKey: []byte(systemPayloadMarkerValue), + "encoding": []byte(encoding), + "messageType": []byte(messageType), }, } attrs := &command.ScheduleNexusOperationCommandAttributes{ - Endpoint: "__temporal_system", + Endpoint: "example-endpoint", Service: "temporal.api.workflowservice.v1.WorkflowService", Operation: "SignalWithStartWorkflowExecution", Input: input, @@ -116,8 +118,8 @@ func decodeEnvelope(t *testing.T, input *common.Payload) *workflowservice.Signal //////////////////////// TESTS //////////////////////// -func TestSystemNexusEnvelopeVisitsInnerPayloads(t *testing.T) { - cmd := buildSystemNexusCommand(t, "binary/protobuf", signalWithStartType) +func TestSystemPayloadVisitsOnlyInnerPayloads(t *testing.T) { + cmd := buildSystemPayloadCommand(t, "binary/protobuf", signalWithStartType) err := VisitPayloads(context.Background(), cmd, VisitPayloadsOptions{ Visitor: collectVisitor, @@ -128,8 +130,8 @@ func TestSystemNexusEnvelopeVisitsInnerPayloads(t *testing.T) { require.True(t, proto.Equal(visitedSignalWithStartRequest, req)) } -func TestSystemNexusEnvelopeVisitsInnerPayloadsConcurrent(t *testing.T) { - cmd := buildSystemNexusCommand(t, "binary/protobuf", signalWithStartType) +func TestSystemPayloadVisitsInnerPayloadsConcurrent(t *testing.T) { + cmd := buildSystemPayloadCommand(t, "binary/protobuf", signalWithStartType) err := VisitPayloads(context.Background(), cmd, VisitPayloadsOptions{ ConcurrencyLimit: 4, @@ -141,8 +143,8 @@ func TestSystemNexusEnvelopeVisitsInnerPayloadsConcurrent(t *testing.T) { require.True(t, proto.Equal(visitedSignalWithStartRequest, req)) } -func TestSystemNexusEnvelopeRejectsNonProtoBinaryEncoding(t *testing.T) { - cmd := buildSystemNexusCommand(t, "json/protobuf", signalWithStartType) +func TestSystemPayloadRejectsNonProtoBinaryEncoding(t *testing.T) { + cmd := buildSystemPayloadCommand(t, "json/protobuf", signalWithStartType) err := VisitPayloads(context.Background(), cmd, VisitPayloadsOptions{ Visitor: trivialVisitor, @@ -151,8 +153,8 @@ func TestSystemNexusEnvelopeRejectsNonProtoBinaryEncoding(t *testing.T) { require.ErrorContains(t, err, "binary/protobuf") } -func TestSystemNexusEnvelopeRejectsUnknownMessageType(t *testing.T) { - cmd := buildSystemNexusCommand(t, "binary/protobuf", "this isn't a valid message type") +func TestSystemPayloadRejectsUnknownMessageType(t *testing.T) { + cmd := buildSystemPayloadCommand(t, "binary/protobuf", "this isn't a valid message type") err := VisitPayloads(context.Background(), cmd, VisitPayloadsOptions{ Visitor: trivialVisitor, @@ -161,8 +163,8 @@ func TestSystemNexusEnvelopeRejectsUnknownMessageType(t *testing.T) { require.ErrorContains(t, err, "unknown message type") } -func TestSystemNexusEnvelopeRejectsMissingMessageType(t *testing.T) { - cmd := buildSystemNexusCommand(t, "binary/protobuf", "") +func TestSystemPayloadRejectsMissingMessageType(t *testing.T) { + cmd := buildSystemPayloadCommand(t, "binary/protobuf", "") err := VisitPayloads(context.Background(), cmd, VisitPayloadsOptions{ Visitor: trivialVisitor, @@ -171,7 +173,7 @@ func TestSystemNexusEnvelopeRejectsMissingMessageType(t *testing.T) { require.ErrorContains(t, err, "missing") } -func TestNonSystemNexusInput(t *testing.T) { +func TestUnmarkedPayloadRetainsExistingBehavior(t *testing.T) { cmd := &command.Command{ Attributes: &command.Command_ScheduleNexusOperationCommandAttributes{ ScheduleNexusOperationCommandAttributes: &command.ScheduleNexusOperationCommandAttributes{ @@ -195,3 +197,40 @@ func TestNonSystemNexusInput(t *testing.T) { require.NoError(t, err) require.Equal(t, []string{"user-payload"}, seen) } + +func TestSystemPayloadInNexusOperationCompletedEvent(t *testing.T) { + data, err := proto.Marshal(signalWithStartRequest) + require.NoError(t, err) + attrs := &history.NexusOperationCompletedEventAttributes{Result: &common.Payload{ + Data: data, + Metadata: map[string][]byte{ + SystemPayloadMetadataKey: []byte(systemPayloadMarkerValue), + "encoding": []byte(binaryProtobufEncoding), + "messageType": []byte(signalWithStartType), + }, + }} + + err = VisitPayloads(context.Background(), attrs, VisitPayloadsOptions{Visitor: collectVisitor}) + require.NoError(t, err) + require.True(t, proto.Equal(visitedSignalWithStartRequest, decodeEnvelope(t, attrs.Result))) +} + +func TestUnmarkedProtobufEnvelopeIsVisitedDirectly(t *testing.T) { + data, err := proto.Marshal(signalWithStartRequest) + require.NoError(t, err) + outer := &common.Payload{Data: data, Metadata: map[string][]byte{ + "encoding": []byte(binaryProtobufEncoding), + "messageType": []byte(signalWithStartType), + }} + attrs := &history.NexusOperationCompletedEventAttributes{Result: outer} + + var visited [][]byte + err = VisitPayloads(context.Background(), attrs, VisitPayloadsOptions{Visitor: func(_ *VisitPayloadsContext, payloads []*common.Payload) ([]*common.Payload, error) { + for _, payload := range payloads { + visited = append(visited, payload.Data) + } + return payloads, nil + }}) + require.NoError(t, err) + require.Equal(t, [][]byte{data}, visited) +} diff --git a/proxy/system_payload.go b/proxy/system_payload.go new file mode 100644 index 00000000..2112392b --- /dev/null +++ b/proxy/system_payload.go @@ -0,0 +1,62 @@ +package proxy + +import ( + "fmt" + + "go.temporal.io/api/common/v1" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoreflect" + "google.golang.org/protobuf/reflect/protoregistry" +) + +// SystemPayloadMetadataKey marks a binary protobuf payload whose data is a +// system envelope containing user payloads that must be visited recursively. +const SystemPayloadMetadataKey = "__temporal_system_payload" + +const ( + systemPayloadMarkerValue = "true" + binaryProtobufEncoding = "binary/protobuf" +) + +// visitSystemPayload unwraps marked system payloads, visits payloads in the +// decoded message, and writes the re-marshaled message back to the envelope. +// It returns true when the payload was handled as a system envelope. +func visitSystemPayload( + ctx *VisitPayloadsContext, + options *VisitPayloadsOptions, + concState *payloadConcurrencyState, + payload *common.Payload, +) (bool, error) { + if payload == nil || string(payload.GetMetadata()[SystemPayloadMetadataKey]) != systemPayloadMarkerValue { + return false, nil + } + + if encoding := string(payload.GetMetadata()["encoding"]); encoding != binaryProtobufEncoding { + return true, fmt.Errorf("system payload must be encoded as %s but got %q", binaryProtobufEncoding, encoding) + } + + messageType := string(payload.GetMetadata()["messageType"]) + if messageType == "" { + return true, fmt.Errorf("system payload is missing the messageType metadata") + } + + messageTypeDescriptor, err := protoregistry.GlobalTypes.FindMessageByName(protoreflect.FullName(messageType)) + if err != nil { + return true, fmt.Errorf("system payload references unknown message type %q: %w", messageType, err) + } + message := messageTypeDescriptor.New().Interface() + if err := proto.Unmarshal(payload.GetData(), message); err != nil { + return true, fmt.Errorf("failed to unmarshal system payload: %w", err) + } + + if err := visitPayloadsAndWait(ctx, options, message, concState, message); err != nil { + return true, err + } + + data, err := proto.Marshal(message) + if err != nil { + return true, fmt.Errorf("failed to marshal system payload: %w", err) + } + payload.Data = data + return true, nil +}