diff --git a/proxy/interceptor_test.go b/proxy/interceptor_test.go index c3d314a1..fbdaae28 100644 --- a/proxy/interceptor_test.go +++ b/proxy/interceptor_test.go @@ -622,392 +622,170 @@ func (t *testGRPCServer) QueryWorkflow( return nil, stWithDetails.Err() } -// Recursively crawl and test Payload(s) with Visitor -func populatePayload(root *proto.Message, msg proto.Message, require *require.Assertions, totalCount *int, count *int) { - m := msg.ProtoReflect() - fields := m.Descriptor().Fields() - // Don't need to parse non-temporal types - if !strings.HasPrefix(string(m.Descriptor().FullName()), "temporal.api.") && string(m.Descriptor().FullName()) != "google.protobuf.Any" { - return - } +// Verify that intercepting a given proto type will result in +// visiting the expected number of commonpb.Payload objects. +func TestVisitedPayloads(t *testing.T) { + // Check that the PayloadWalker works as expected. + t.Run("Walking", func(t *testing.T) { + w := RunPayloadWalker(t, "temporal.api.nexus.v1.Request") + w.RequirePayloadCount(1) + // The six messages visited are nexus.v1.Request itself plus: + // - Request.Capabilities + // - StartOperationRequest (via the variant oneof) + // - CancelOperationRequest (via the variant oneof) + // - commonpb.Payload (StartOperationRequest.payload) + // - nexus.v1.Link (StartOperationRequest.links) + // + // References to google.protobuf.Timestamp aren't counted, since + // they aren't Temporal messages. + require.Equal(t, 6, w.MessagesVisited) + }) - if m.Descriptor() == nil { - panic("fail") + tests := []struct { + protoType string + wantPayloadCount int + }{ + {"temporal.api.update.v1.Rejection", 7}, + {"temporal.api.query.v1.WorkflowQueryResult", 6}, + {"temporal.api.protocol.v1.Message", 1}, + {"temporal.api.command.v1.Command", 38}, + + // repeated temporal.api.command.v1.Command commands - 38 + // map query_results - 6 + // repeated temporal.api.protocol.v1.Message messages - 1 + // TOTAL - 45 + {"temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedRequest", 45}, + + {"temporal.api.workflowservice.v1.CountWorkflowExecutionsResponse", 1}, + {"temporal.api.update.v1.Response", 6}, + {"temporal.api.errordetails.v1.MultiOperationExecutionFailure.OperationStatus", 1}, } - - // Base case, ensure Visitor can reach Payload from root Message - switch i := msg.(type) { - case *common.Payload, *common.Payloads: - *count++ - *totalCount++ - err := VisitPayloads(context.Background(), *root, VisitPayloadsOptions{ - Visitor: func(ctx *VisitPayloadsContext, p []*common.Payload) ([]*common.Payload, error) { - require.Equal(1, *count) - *count-- - return p, nil - }, + for _, test := range tests { + t.Run(test.protoType, func(t *testing.T) { + w := RunPayloadWalker(t, test.protoType) + w.RequirePayloadCount(test.wantPayloadCount) }) - require.NoError(err) - return - case *anypb.Any: - if i.TypeUrl == "" { - // Set to a random proto struct we know contains a payload, to test if we - // are able to recurse through Any to reach a payload - newAny, err := anypb.New(&update.Request{Input: &update.Input{Args: &common.Payloads{ - Payloads: []*common.Payload{{Data: []byte("orig-val")}}, - }}}) - require.NoError(err) - proto.Merge(msg, newAny) - } - *count++ - *totalCount++ - err := VisitPayloads(context.Background(), *root, VisitPayloadsOptions{ - Visitor: func(ctx *VisitPayloadsContext, p []*common.Payload) ([]*common.Payload, error) { - require.Equal(1, *count) - *count-- - return p, nil - }, - }) - require.NoError(err) - return - } - - // Go through all fields, populating each then recursing into them to discover Payloads to test - // with Visitor - for i := 0; i < fields.Len(); i++ { - fd := fields.Get(i) - value := m.Get(fd) - - if oneof := fd.ContainingOneof(); oneof != nil && fd.Kind() == protoreflect.MessageKind { - newMsg := value.Message().New() - m.Set(fd, protoreflect.ValueOf(newMsg)) - populatePayload(root, newMsg.Interface(), require, totalCount, count) - // This ensures only 1 payload is set and discoverable from root at a time. - m.Clear(fd) - } else if fd.IsMap() { - mapVal := m.Mutable(fd).Map() - require.Equal(0, mapVal.Len()) - if fd.MapKey().Kind() == protoreflect.StringKind && - fd.MapValue().Kind() == protoreflect.MessageKind && - string(fd.MapValue().Message().FullName()) == "temporal.api.common.v1.Payload" { - sampleKey := protoreflect.ValueOf("sample_key").MapKey() - mapVal.Set(sampleKey, protoreflect.ValueOf(inputPayload().ProtoReflect())) - mapVal.Range(func(key protoreflect.MapKey, val protoreflect.Value) bool { - if fd.MapValue().Kind() == protoreflect.MessageKind { - populatePayload(root, val.Message().Interface(), require, totalCount, count) - } - return true - }) - mapVal.Clear(sampleKey) - } else if fd.MapValue().Kind() == protoreflect.MessageKind { - var sampleKey protoreflect.MapKey - switch fd.MapKey().Kind() { - case protoreflect.StringKind: - sampleKey = protoreflect.ValueOf("sample_key").MapKey() - case protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Sfixed32Kind: - sampleKey = protoreflect.ValueOf(int32(1)).MapKey() - case protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Uint64Kind: - sampleKey = protoreflect.ValueOf(int64(1)).MapKey() - case protoreflect.BoolKind: - sampleKey = protoreflect.ValueOf(true).MapKey() - default: - fmt.Println("Skipping unsupported map key type:", fd.MapKey().Kind()) - return - } - mapVal.Set(sampleKey, mapVal.NewValue()) - mapVal.Range(func(key protoreflect.MapKey, val protoreflect.Value) bool { - if fd.MapValue().Kind() == protoreflect.MessageKind { - newMsg := val.Message() - populatePayload(root, newMsg.Interface(), require, totalCount, count) - } - return true - }) - // This ensures only 1 payload is set and discoverable from root at a time. - mapVal.Clear(sampleKey) - } - } else if fd.IsList() { - if fd.Kind() == protoreflect.MessageKind { - listVal := m.Mutable(fd).List() - require.Equal(0, listVal.Len()) - - sampleVal := listVal.NewElement() - listVal.Append(sampleVal) - - val := listVal.Get(0) - require.True(val.Message().IsValid()) - require.Equal(1, listVal.Len()) - populatePayload(root, sampleVal.Message().Interface(), require, totalCount, count) - // This ensures only 1 payload is set and discoverable from root at a time. - listVal.Truncate(0) - } - } else { - if fd.Kind() == protoreflect.MessageKind { - // Avoid cycles - if value.Message().Descriptor().FullName() == m.Descriptor().FullName() { - continue - } - - var newMsg protoreflect.Message - newMsg = value.Message().New() - m.Set(fd, protoreflect.ValueOf(newMsg)) - populatePayload(root, newMsg.Interface(), require, totalCount, count) - // This ensures only 1 payload is set and discoverable from root at a time. - m.Clear(fd) - } - } - // Validate that all Payloads were found - require.Equal(0, *count) } } -func TestVisitPayloads_FailureCount(t *testing.T) { - require := require.New(t) - - var messageType protoreflect.MessageType +func TestVisitPayloads_Everything(t *testing.T) { + var messageTypes []protoreflect.MessageType protoregistry.GlobalTypes.RangeMessages(func(mt protoreflect.MessageType) bool { - if strings.HasPrefix(string(mt.Descriptor().FullName()), "temporal.api.failure.v1.Failure") { - messageType = mt + fullName := string(mt.Descriptor().FullName()) + // The base case of passing Payload into the visitor is not supported. + // See godoc for VisitPayloads + if strings.HasPrefix(fullName, temporalPrefix) && fullName != payloadFullName { + messageTypes = append(messageTypes, mt) } return true }) + require.NotEmpty(t, messageTypes) - // Create empty instance and populate with test values - msg := messageType.New().Interface().(proto.Message) - var totalCount, count int - populatePayload(&msg, msg, require, &totalCount, &count) + for _, mt := range messageTypes { + fullName := string(mt.Descriptor().FullName()) - require.Equal(0, count) - require.Equal(5, totalCount) + // Visit every Temporal payload we see. This ensures that we don't + // panic or fail on some exotic proto types. + RunPayloadWalker(t, fullName) + } } -func TestVisitPayloads_UpdateRejectionCount(t *testing.T) { - require := require.New(t) - - var messageType protoreflect.MessageType - protoregistry.GlobalTypes.RangeMessages(func(mt protoreflect.MessageType) bool { - if strings.HasPrefix(string(mt.Descriptor().FullName()), "temporal.api.update.v1.Rejection") { - messageType = mt - } - return true - }) - - // Create empty instance and populate with test values - msg := messageType.New().Interface().(proto.Message) - var totalCount, count int - populatePayload(&msg, msg, require, &totalCount, &count) - - require.Equal(0, count) - require.Equal(7, totalCount) +type fakePayloadVisitor struct { + payloadsEncountered int + anysEncountered int } -func TestVisitPayloads_PayloadsCount(t *testing.T) { - require := require.New(t) - - var messageType protoreflect.MessageType - protoregistry.GlobalTypes.RangeMessages(func(mt protoreflect.MessageType) bool { - if strings.HasPrefix(string(mt.Descriptor().FullName()), "temporal.api.query.v1.WorkflowQueryResult") { - messageType = mt - } - return true - }) - - // Create empty instance and populate with test values - msg := messageType.New().Interface().(proto.Message) - var totalCount, count int - populatePayload(&msg, msg, require, &totalCount, &count) - - require.Equal(0, count) - require.Equal(6, totalCount) +func (fpv *fakePayloadVisitor) ToVisitPayloadsOptions() VisitPayloadsOptions { + return VisitPayloadsOptions{ + Visitor: func(_ *VisitPayloadsContext, p []*common.Payload) ([]*common.Payload, error) { + fpv.payloadsEncountered++ + return p, nil + }, + WellKnownAnyVisitor: func(_ *VisitPayloadsContext, _ *anypb.Any) error { + fpv.anysEncountered++ + return nil + }, + } } -func TestVisitPayloads_AnyCount(t *testing.T) { - require := require.New(t) - - var messageType protoreflect.MessageType - protoregistry.GlobalTypes.RangeMessages(func(mt protoreflect.MessageType) bool { - if string(mt.Descriptor().FullName()) == "temporal.api.protocol.v1.Message" { - messageType = mt - } - return true +// The RunPayloadWalker test harness skips self-referential types to avoid +// blowing up when there are cycles in the proto types. But there are legitimate +// cases where VisitPayloads(...) is called with a self-referential type +// and so we test the behavior here. +// +// BUG: If there was a cycle in the actual proto graph, we would expect +// VisitPayloads to stack overflow. +func TestVisitPayloads_SelfReferentialTypes(t *testing.T) { + t.Run("ViaRunPayloadWalker", func(t *testing.T) { + w := RunPayloadWalker(t, "temporal.api.failure.v1.Failure") + // The test harness confirms that there are 5x Payloads/Any + // types in the proto graph. + w.RequirePayloadCount(5) }) - // Create empty instance and populate with test values - msg := messageType.New().Interface().(proto.Message) - var totalCount, count int - populatePayload(&msg, msg, require, &totalCount, &count) - - require.Equal(0, count) - require.Equal(1, totalCount) -} - -func TestVisitPayloads_CommandCount(t *testing.T) { - require := require.New(t) - // UserMetadata - 2 - // ScheduleActivityTaskCommandAttributes - 2 - // header - 1 - // payloads - 1 - // CompleteWorkflowExecutionCommandAttributes - 1 - // FailWorkflowExecutionCommandAttributes - 5 - // failure - 5 - // CancelWorkflowExecutionCommandAttributes - 1 - // RecordMarkerCommandAttributes - 7 - // details - 1 - // header - 1 - // failure - 5 - // ContinueAsNewWorkflowExecutionCommandAttributes - 10 - // input - 1 - // failure - 5 - // last_completion_result - 1 - // header - 1 - // memo - 1 - // SearchAttributes - 1 - // - // StartChildWorkflowExecutionCommandAttributes - 4 - // input - 1 - // header - // memo - // searchAttributes - // SignalExternalWorkflowExecutionCommandAttributes - 2 - // header - 1 - // input - 1 - // UpsertWorkflowSearchAttributesCommandAttributes - 1 - // searchAttributes - 1 - // ModifyWorkflowPropertiesCommandAttributes - 1 - // memo - 1 - // ScheduleNexusOperationCommandAttributes - 1 - // input - 1 - // EventGroupMarkers (Command.event_group_markers) - 1 - // label.label - 1 - // TOTAL : 38 - var messageType protoreflect.MessageType - protoregistry.GlobalTypes.RangeMessages(func(mt protoreflect.MessageType) bool { - if string(mt.Descriptor().FullName()) == "temporal.api.command.v1.Command" { - messageType = mt + t.Run("NestedFailures", func(t *testing.T) { + // Create some stock failure_info values, with and without nested Payloads. + fiWithPayload1 := &failure.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failure.ApplicationFailureInfo{ + Details: &common.Payloads{}, + }, } - return true - }) - - // Create empty instance and populate with test values - msg := messageType.New().Interface().(proto.Message) - var totalCount, count int - populatePayload(&msg, msg, require, &totalCount, &count) - - require.Equal(0, count) - require.Equal(38, totalCount) -} - -func TestVisitPayloads_MapCount(t *testing.T) { - require := require.New(t) - - var messageType protoreflect.MessageType - var totalCount, count int - - // repeated temporal.api.command.v1.Command commands - 38 - // map query_results - 6 - // repeated temporal.api.protocol.v1.Message messages - 1 - // TOTAL - 45 - protoregistry.GlobalTypes.RangeMessages(func(mt protoreflect.MessageType) bool { - if string(mt.Descriptor().FullName()) == "temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedRequest" { - messageType = mt + fiWithPayload2 := &failure.Failure_TimeoutFailureInfo{ + TimeoutFailureInfo: &failure.TimeoutFailureInfo{ + LastHeartbeatDetails: &common.Payloads{}, + }, } - return true - }) - - // Create empty instance and populate with test values - msg1 := messageType.New().Interface().(proto.Message) - totalCount = 0 - count = 0 - populatePayload(&msg1, msg1, require, &totalCount, &count) - - require.Equal(0, count) - require.Equal(45, totalCount) -} - -func TestVisitPayloads_CountWorkflowExecutionsResponse(t *testing.T) { - require := require.New(t) - - var messageType protoreflect.MessageType - var totalCount, count int - - protoregistry.GlobalTypes.RangeMessages(func(mt protoreflect.MessageType) bool { - if string(mt.Descriptor().FullName()) == "temporal.api.workflowservice.v1.CountWorkflowExecutionsResponse" { - messageType = mt + fiWithoutPayload1 := &failure.Failure_ActivityFailureInfo{ + ActivityFailureInfo: &failure.ActivityFailureInfo{ + // Type contains no Payloads. + }, } - return true - }) - - // Create empty instance and populate with test values - msg1 := messageType.New().Interface().(proto.Message) - totalCount = 0 - count = 0 - populatePayload(&msg1, msg1, require, &totalCount, &count) - - require.Equal(0, count) - require.Equal(1, totalCount) -} - -func TestVisitPayloads_ResponseCount(t *testing.T) { - require := require.New(t) - - var messageType protoreflect.MessageType - protoregistry.GlobalTypes.RangeMessages(func(mt protoreflect.MessageType) bool { - if string(mt.Descriptor().FullName()) == "temporal.api.update.v1.Response" { - messageType = mt + fiWithoutPayload2 := &failure.Failure_NexusHandlerFailureInfo{ + NexusHandlerFailureInfo: &failure.NexusHandlerFailureInfo{ + // Type contains no Payloads. + }, } - return true - }) - // Create empty instance and populate with test values - msg := messageType.New().Interface().(proto.Message) - var totalCount, count int - populatePayload(&msg, msg, require, &totalCount, &count) - - require.Equal(0, count) - require.Equal(6, totalCount) -} - -func TestVisitPayloads_OperationStatus(t *testing.T) { - require := require.New(t) - - var messageType protoreflect.MessageType - protoregistry.GlobalTypes.RangeMessages(func(mt protoreflect.MessageType) bool { - if string(mt.Descriptor().FullName()) == "temporal.api.errordetails.v1.MultiOperationExecutionFailure.OperationStatus" { - messageType = mt + // Construct a failurepb.Failure that nests a payload in one + // of the self-referential fields that RunPayloadWalker omits. + // + // - We nest 3x more failure.Failure protos via the cause field. + // - Each failure has a failure_info, which may or may not contain + // a Payload. + // - We set the encoded_attributes for some, adding a Payload. + trickyFailureObj := failure.Failure{ + Message: "failure-1", + FailureInfo: fiWithPayload1, + EncodedAttributes: &common.Payload{}, // +1 payload + Cause: &failure.Failure{ + Message: "failure-2", + FailureInfo: fiWithoutPayload1, + Cause: &failure.Failure{ + Message: "failure-3", + FailureInfo: fiWithPayload2, + Cause: &failure.Failure{ + Message: "failure-4", + FailureInfo: fiWithoutPayload2, + EncodedAttributes: &common.Payload{}, // +1 payload + Cause: nil, + }, + }, + }, } - return true - }) - // Create empty instance and populate with test values - msg := messageType.New().Interface().(proto.Message) - var totalCount, count int - populatePayload(&msg, msg, require, &totalCount, &count) - - require.Equal(0, count) - require.Equal(1, totalCount) -} + // Create fake visitors that just count the number of payloads + // and well known Any types they see. + var visitor fakePayloadVisitor -func TestVisitPayloads_Everything(t *testing.T) { - require := require.New(t) + ctx := context.Background() + err := VisitPayloads(ctx, &trickyFailureObj, visitor.ToVisitPayloadsOptions()) + require.NoError(t, err) - var messageType []protoreflect.MessageType - protoregistry.GlobalTypes.RangeMessages(func(mt protoreflect.MessageType) bool { - // The base case of passing Payload into the visitor is not supported. - // See godoc for VisitPayloads - if strings.HasPrefix(string(mt.Descriptor().FullName()), "temporal.api.") && string(mt.Descriptor().FullName()) != "temporal.api.common.v1.Payload" { - messageType = append(messageType, mt) - } - return true + // 2x Payload from the encoded_attributes. + // 2x Payloads from failure_infos (fiWithPayload1, fiWithPayload2). + require.Equal(t, 4, visitor.payloadsEncountered, "unexpected number of Payloads encountered") + require.Equal(t, 0, visitor.anysEncountered, "unexpected number of Any types encountered") }) - for _, mt := range messageType { - // Create empty instance and populate with test values - msg := mt.New().Interface().(proto.Message) - - var totalCount, count int - populatePayload(&msg, msg, require, &totalCount, &count) - - require.Equal(0, count) - - } } // contextHookKey is an unexported key for values injected by ContextHook tests. diff --git a/proxy/payload_walker_test.go b/proxy/payload_walker_test.go new file mode 100644 index 00000000..ac6bace4 --- /dev/null +++ b/proxy/payload_walker_test.go @@ -0,0 +1,308 @@ +package proxy + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/require" + + "go.temporal.io/api/common/v1" + "go.temporal.io/api/update/v1" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoreflect" + "google.golang.org/protobuf/reflect/protoregistry" + "google.golang.org/protobuf/types/known/anypb" +) + +const ( + payloadFullName = "temporal.api.common.v1.Payload" + payloadsFullName = "temporal.api.common.v1.Payloads" + anyFullName = "google.protobuf.Any" + temporalPrefix = "temporal.api." + + // maxWalkDepth bounds how far the walker will descend. If there is a cycle + // (e.g. A holds a B, and a B holds an A) we hit this max depth and fail + // rather than panic with a stack overflow. + maxWalkDepth = 50 +) + +// PayloadWalker verifies that VisitPayloads can reach every payload that a proto +// message tree is capable of holding, no matter how deeply it is nested. +// +// It works by walking the descriptors of a root message and populating exactly +// one leaf at a time: for each field it creates a fresh child message, descends +// into it, then clears the field again. Whenever the walk reaches a payload +// container (Payload, Payloads, or an Any stuffed with a payload-bearing +// message) it runs VisitPayloads against the *root* message and asserts the +// Visitor fires exactly once, proving the generated visitor code has a path +// from the root all the way down to that leaf. +// +// The zero value is not usable; construct one with RunPayloadWalker. +type PayloadWalker struct { + t *testing.T + + // MessagesVisited is the number of proto messages the walker descended into. + // Non-Temporal messages (other than Any) are not walked and not counted. + MessagesVisited int + + // PayloadsSeen has one entry per payload container the walker reached, in + // discovery order, formatted as " []". + PayloadsSeen []string + + // Skipped has one entry per subtree the walker declined to enter, along with + // the reason. (e.g. self-referential fields such as Failure.cause would + //recurse forever.) + Skipped []string + + root proto.Message + // rootName is the full name of the message the walk started from. Kept + // separately from path[0] so that Report stays correct no matter where in the + // walk it is called from. + rootName string + + // path is the field path from root down to the message currently being walked. + path []string + + // pendingVisits counts payload containers that have been populated and + // handed to VisitPayloads but that the Visitor has not reported back yet. + // Because only one payload is populated at a time it is 1 for the duration of + // a single VisitPayloads call and 0 everywhere else; any other value means + // the Visitor fired the wrong number of times. + pendingVisits int +} + +// RunPayloadWalker loads the named protobuf message and recursively walks its +// fields, recording the results in the returned PayloadWalker. +func RunPayloadWalker(t *testing.T, fullName string) *PayloadWalker { + t.Helper() + + root := mustGetProtoByName(t, fullName) + w := &PayloadWalker{ + t: t, + root: root, + rootName: fullName, + path: []string{fullName}, + } + w.walk(root.ProtoReflect()) + require.Equal(t, 0, w.pendingVisits, "walk finished with unvisited payloads:\n%s", w.Report()) + return w +} + +// PayloadCount is the number of payload containers the walk reached. +func (w *PayloadWalker) PayloadCount() int { return len(w.PayloadsSeen) } + +// RequirePayloadCount asserts the walk found exactly want payload containers, +// printing every payload it did find when it did not. Prefer this over +// comparing PayloadCount by hand for better error messages. +func (w *PayloadWalker) RequirePayloadCount(want int) { + w.t.Helper() + + // Hello! Test failing? That doesn't necessarily mean you did something + // wrong. If you have added/removed a proto field then the hard-coded number + // of Payloads we look for in a test might have changed. + // + // Double check that is the case (e.g. w.Report() includes a new proto field + // you added). + require.Equal( + w.t, want, w.PayloadCount(), + "expected %d payloads reachable from %s, found %d\n%s", + want, w.rootName, w.PayloadCount(), w.Report()) +} + +// Report renders the walk as a human readable summary. +func (w *PayloadWalker) Report() string { + var sb strings.Builder + fmt.Fprintf(&sb, "%s: %d messages visited, %d payloads seen\n", + w.rootName, w.MessagesVisited, w.PayloadCount()) + sb.WriteString("payloads seen:\n") + + for i, p := range w.PayloadsSeen { + fmt.Fprintf(&sb, " %3d. %s\n", i+1, p) + } + + if len(w.Skipped) > 0 { + sb.WriteString("subtrees skipped:\n") + for _, s := range w.Skipped { + fmt.Fprintf(&sb, " - %s\n", s) + } + } + return sb.String() +} + +// walk descends into a single message, populating one field at a time. +func (w *PayloadWalker) walk(m protoreflect.Message) { + fullName := string(m.Descriptor().FullName()) + + // Non-Temporal messages cannot contain payloads. The `Any` type is the exception, + // it can wrap one, and the visitor is expected to recurse through it. + if !strings.HasPrefix(fullName, temporalPrefix) && fullName != anyFullName { + return + } + require.LessOrEqual(w.t, len(w.path), maxWalkDepth, + "walk exceeded a depth of %d, which suggests a reference cycle in the protos at %s", + maxWalkDepth, w.pathString()) + w.MessagesVisited++ + + // Base case: this message is (or, for Any, can be made to hold) a payload + // container, so stop descending and check the visitor reaches it. + switch fullName { + case payloadFullName, payloadsFullName: + w.visitFromRoot(fullName) + return + + case anyFullName: + wrapper, ok := m.Interface().(*anypb.Any) + require.True(w.t, ok, "expected an *anypb.Any at %s, got %T", w.pathString(), m.Interface()) + if wrapper.TypeUrl == "" { + // Stuff the Any with an arbitrary message we know holds a payload, so + // that reaching this payload requires recursing through the Any. + stuffed, err := anypb.New(&update.Request{Input: &update.Input{Args: &common.Payloads{ + Payloads: []*common.Payload{{Data: []byte("orig-val")}}, + }}}) + require.NoError(w.t, err) + proto.Merge(wrapper, stuffed) + } + w.visitFromRoot(fullName) + return + } + + fields := m.Descriptor().Fields() + for i := 0; i < fields.Len(); i++ { + fd := fields.Get(i) + w.walkField(m, fd) + // Everything populated under this field has been cleared again, so any + // payload announced while descending must have been reported back by now. + require.Equal(w.t, 0, w.pendingVisits, + "VisitPayloads did not reach the payload under %s.%s", w.pathString(), fd.Name()) + } +} + +// walkField populates fd on m, descends into whatever it created, then clears it +// so that only one payload is ever reachable from the root at a time. +func (w *PayloadWalker) walkField(m protoreflect.Message, fd protoreflect.FieldDescriptor) { + w.path = append(w.path, string(fd.Name())) + defer func() { w.path = w.path[:len(w.path)-1] }() + + switch { + case fd.IsMap(): + w.walkMapField(m, fd) + case fd.IsList(): + w.walkListField(m, fd) + case fd.Kind() == protoreflect.MessageKind: + w.walkMessageField(m, fd) + } + // Scalar fields cannot hold payloads, so there is nothing to populate. +} + +// walkMessageField handles a singular message field, including the members of a +// oneof (each oneof member is populated and cleared in turn). +func (w *PayloadWalker) walkMessageField(m protoreflect.Message, fd protoreflect.FieldDescriptor) { + // A field of the same type as its parent (e.g. Failure.cause) would recurse + // forever. Payloads underneath it are reachable through the parent anyway. + if fd.Message().FullName() == m.Descriptor().FullName() { + w.skip("self-referential field") + return + } + + child := m.Get(fd).Message().New() + m.Set(fd, protoreflect.ValueOfMessage(child)) + w.walk(child) + m.Clear(fd) +} + +// walkListField handles a repeated field by appending a single element. +func (w *PayloadWalker) walkListField(m protoreflect.Message, fd protoreflect.FieldDescriptor) { + if fd.Kind() != protoreflect.MessageKind { + return + } + list := m.Mutable(fd).List() + require.Equal(w.t, 0, list.Len(), "walker expects an empty list at %s", w.pathString()) + + elem := list.NewElement() + list.Append(elem) + w.walk(elem.Message()) + list.Truncate(0) +} + +// walkMapField handles a map field by setting a single entry. +func (w *PayloadWalker) walkMapField(m protoreflect.Message, fd protoreflect.FieldDescriptor) { + if fd.MapValue().Kind() != protoreflect.MessageKind { + return + } + mapVal := m.Mutable(fd).Map() + require.Equal(w.t, 0, mapVal.Len(), "walker expects an empty map at %s", w.pathString()) + + key := mustGetMapKey(w.t, fd.MapKey().Kind()) + + // For map<_, Payload> fields, populate real payload data rather than an empty + // message, matching how callers actually use them (e.g. Header.fields). + entry := mapVal.NewValue() + if fd.MapValue().Message().FullName() == payloadFullName { + entry = protoreflect.ValueOfMessage(inputPayload().ProtoReflect()) + } + mapVal.Set(key, entry) + w.walk(entry.Message()) + mapVal.Clear(key) +} + +// visitFromRoot records a payload container at the current path and asserts that +// VisitPayloads, run against the root message, invokes the Visitor exactly once. +// +// Exactly one payload is populated at this point, so exactly one call is +// expected: zero calls means the visitor cannot reach this field, more than one +// means it visits it repeatedly. +func (w *PayloadWalker) visitFromRoot(leafType string) { + w.PayloadsSeen = append(w.PayloadsSeen, fmt.Sprintf("%s [%s]", w.pathString(), leafType)) + + w.pendingVisits++ + + ctx := context.Background() + err := VisitPayloads(ctx, w.root, VisitPayloadsOptions{ + Visitor: func(ctx *VisitPayloadsContext, p []*common.Payload) ([]*common.Payload, error) { + require.Equal(w.t, 1, w.pendingVisits, + "VisitPayloads invoked the Visitor more than once for %s", w.pathString()) + w.pendingVisits-- + return p, nil + }, + }) + require.NoError(w.t, err) +} + +func (w *PayloadWalker) skip(reason string) { + w.Skipped = append(w.Skipped, fmt.Sprintf("%s (%s)", w.pathString(), reason)) +} + +func (w *PayloadWalker) pathString() string { + return strings.Join(w.path, ".") +} + +// mustGetMapKey returns an arbitrary map key of the given kind. Panics +// if the kind is not supported. +func mustGetMapKey(t *testing.T, kind protoreflect.Kind) protoreflect.MapKey { + switch kind { + case protoreflect.StringKind: + return protoreflect.ValueOfString("sample_key").MapKey() + case protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Sfixed32Kind: + return protoreflect.ValueOfInt32(1).MapKey() + case protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Sfixed64Kind: + return protoreflect.ValueOfInt64(1).MapKey() + case protoreflect.Uint64Kind: + return protoreflect.ValueOfUint64(1).MapKey() + case protoreflect.BoolKind: + return protoreflect.ValueOfBool(true).MapKey() + } + + t.Fatalf("Map key kind %v not implemented. Please add support for it.", kind) + return protoreflect.MapKey{} +} + +// mustGetProtoByName fetches the proto message from its qualified name, +// returning its proto.Message interface. +func mustGetProtoByName(t *testing.T, fullName string) proto.Message { + t.Helper() + mt, err := protoregistry.GlobalTypes.FindMessageByName(protoreflect.FullName(fullName)) + require.NoError(t, err, "message %s not found in the global proto registry", fullName) + return mt.New().Interface() +}