grpc: accept legacy stream protocol clients via Meta("grpc:stream:compat", "v1")

dsl/meta.go

@@ -187,6 +187,25 @@ const DefaultProtoc = expr.DefaultProtoc
 //	    Meta("protoc:include", "/usr/local/include/google/protobuf")
 //	})
 //
+// - "grpc:stream:compat" makes generated gRPC servers also accept clients
+// generated by goa versions that predate the typed stream envelope for
+// methods that define both Payload and StreamingPayload. Such legacy clients
+// carry the one-shot method payload in gRPC request metadata instead of a
+// typed initial stream frame. The only supported value is "v1". Applicable
+// to API, service and method definitions; the legacy protocol only encodes
+// primitives and arrays of primitives so the method payload must be limited
+// to such attributes. Remove the meta once all clients are upgraded to drop
+// the compatibility code.
+//
+//	var _ = Service("MyService", func() {
+//	    Method("MyMethod", func() {
+//	        Meta("grpc:stream:compat", "v1")
+//	        Payload(MyPayload)
+//	        StreamingPayload(MyStreamItem)
+//	        GRPC(func() {})
+//	    })
+//	})
+//
 // - "swagger:generate" DEPRECATED, use "openapi:generate" instead.
 //
 // - "openapi:generate" specifies whether OpenAPI specification should be

expr/grpc_endpoint.go

@@ -35,6 +35,16 @@ type (
 	}
 )
 
+const (
+	// streamCompatMetaKey is the meta key that makes generated servers also
+	// accept clients speaking the legacy stream protocol which predates the
+	// typed stream envelope.
+	streamCompatMetaKey = "grpc:stream:compat"
+	// streamCompatLegacy is the only supported value of the stream
+	// compatibility meta and selects the legacy metadata-based protocol.
+	streamCompatLegacy = "v1"
+)
+
 // Name of gRPC endpoint
 func (e *GRPCEndpointExpr) Name() string {
 	return e.MethodExpr.Name
@@ -143,6 +153,16 @@ func (e *GRPCEndpointExpr) Prepare() {
 	}
 }
 
+// LegacyStreamCompat reports whether the generated server must also accept
+// clients that speak the legacy stream protocol which carries the one-shot
+// method payload in gRPC request metadata instead of a typed initial stream
+// frame. It is enabled by setting Meta("grpc:stream:compat", "v1") on the
+// method, the service or the API.
+func (e *GRPCEndpointExpr) LegacyStreamCompat() bool {
+	v, ok := e.streamCompatValue()
+	return ok && v == streamCompatLegacy
+}
+
 // Validate validates the endpoint expression by checking if the request
 // and responses contains the "rpc:tag" in the meta. It also makes sure
 // that there is only one response per status code.
@@ -151,6 +171,7 @@ func (e *GRPCEndpointExpr) Validate() error {
 	if e.Name() == "" {
 		verr.Add(e, "Endpoint name cannot be empty")
 	}
+	verr.Merge(e.validateStreamCompat())
 
 	seenUnions := make(map[*Union]struct{})
 	seenAttrs := make(map[*AttributeExpr]struct{})
@@ -568,3 +589,75 @@ func getSecurityAttributes(m *MethodExpr) []string {
 	}
 	return secAttrs
 }
+
+// streamCompatValue returns the value of the stream compatibility meta by
+// looking up the endpoint, method, service and API expressions in that order.
+func (e *GRPCEndpointExpr) streamCompatValue() (string, bool) {
+	if v, ok := e.Meta.Last(streamCompatMetaKey); ok {
+		return v, true
+	}
+	if v, ok := e.MethodExpr.Meta.Last(streamCompatMetaKey); ok {
+		return v, true
+	}
+	if v, ok := e.Service.ServiceExpr.Meta.Last(streamCompatMetaKey); ok {
+		return v, true
+	}
+	if v, ok := Root.API.Meta.Last(streamCompatMetaKey); ok {
+		return v, true
+	}
+	return "", false
+}
+
+// validateStreamCompat validates the stream compatibility meta if set. The
+// legacy stream protocol carries the one-shot method payload in gRPC metadata
+// which can only encode primitive values and arrays of primitive values.
+func (e *GRPCEndpointExpr) validateStreamCompat() *eval.ValidationErrors {
+	verr := new(eval.ValidationErrors)
+	value, ok := e.streamCompatValue()
+	if !ok {
+		return verr
+	}
+	if value != streamCompatLegacy {
+		verr.Add(e, "invalid %q meta value %q: only %q is supported", streamCompatMetaKey, value, streamCompatLegacy)
+		return verr
+	}
+	if !e.MethodExpr.IsPayloadStreaming() || e.MethodExpr.Payload.Type == Empty {
+		// The meta only affects methods that combine a one-shot payload with
+		// a streaming payload. Only report a missing effect when the meta is
+		// set on the endpoint or method directly; service and API level metas
+		// legitimately apply to a subset of the methods they cover.
+		_, endpointLevel := e.Meta.Last(streamCompatMetaKey)
+		_, methodLevel := e.MethodExpr.Meta.Last(streamCompatMetaKey)
+		if endpointLevel || methodLevel {
+			verr.Add(e, "%q meta requires the method to define both Payload and StreamingPayload", streamCompatMetaKey)
+		}
+		return verr
+	}
+	if obj := AsObject(e.MethodExpr.Payload.Type); obj != nil {
+		metObj := AsObject(e.Metadata.Type)
+		for _, nat := range *obj {
+			if metObj.Attribute(nat.Name) != nil {
+				// Attributes explicitly mapped to metadata travel in metadata
+				// under both protocols and are validated separately.
+				continue
+			}
+			if !isMetadataEncodable(nat.Attribute.Type) {
+				verr.Add(e, "attribute %q of the method payload must be a primitive or an array of primitives to satisfy the %q meta", nat.Name, streamCompatMetaKey)
+			}
+		}
+	} else if !isMetadataEncodable(e.MethodExpr.Payload.Type) {
+		verr.Add(e, "the method payload must be a primitive, an array of primitives or an object to satisfy the %q meta", streamCompatMetaKey)
+	}
+	return verr
+}
+
+// isMetadataEncodable reports whether values of the given type can be carried
+// in gRPC metadata, that is whether they can be encoded to and decoded from
+// header strings.
+func isMetadataEncodable(dt DataType) bool {
+	if IsPrimitive(dt) {
+		return true
+	}
+	arr := AsArray(dt)
+	return arr != nil && IsPrimitive(arr.ElemType.Type)
+}

expr/grpc_endpoint_test.go

@@ -44,6 +44,26 @@ service "Service" gRPC endpoint "Method": field number 2 in attribute "key_dup_i
 			DSL:    testdata.GRPCEndpointWithInheritErrorDSL,
 			Errors: []string{},
 		},
+		"endpoint-stream-compat": {
+			DSL:    testdata.GRPCEndpointStreamCompat,
+			Errors: []string{},
+		},
+		"endpoint-stream-compat-service-level": {
+			DSL:    testdata.GRPCEndpointStreamCompatServiceLevel,
+			Errors: []string{},
+		},
+		"endpoint-stream-compat-bad-value": {
+			DSL:    testdata.GRPCEndpointStreamCompatBadValue,
+			Errors: []string{`service "Service" gRPC endpoint "Method": invalid "grpc:stream:compat" meta value "v2": only "v1" is supported`},
+		},
+		"endpoint-stream-compat-no-streaming-payload": {
+			DSL:    testdata.GRPCEndpointStreamCompatNoStreamingPayload,
+			Errors: []string{`service "Service" gRPC endpoint "Method": "grpc:stream:compat" meta requires the method to define both Payload and StreamingPayload`},
+		},
+		"endpoint-stream-compat-union-payload": {
+			DSL:    testdata.GRPCEndpointStreamCompatUnionPayload,
+			Errors: []string{`service "Service" gRPC endpoint "Method": attribute "version_ref" of the method payload must be a primitive or an array of primitives to satisfy the "grpc:stream:compat" meta`},
+		},
 		"endpoint-union-containing-any": {
 			DSL: testdata.GRPCEndpointWithUnionContainingAny,
 			Errors: []string{
@@ -100,3 +120,24 @@ func TestGRPCEndpointStreamingPayloadKeepsInitialRequest(t *testing.T) {
 	require.NotNil(t, req.Attribute("version_ref"))
 	require.True(t, endpoint.Metadata.IsEmpty())
 }
+
+func TestGRPCEndpointLegacyStreamCompat(t *testing.T) {
+	cases := []struct {
+		Name     string
+		DSL      func()
+		Expected bool
+	}{
+		{"method-level", testdata.GRPCEndpointStreamCompat, true},
+		{"service-level", testdata.GRPCEndpointStreamCompatServiceLevel, true},
+		{"not-set", testdata.GRPCEndpointWithStreamingPayloadInitialRequest, false},
+	}
+	for _, c := range cases {
+		t.Run(c.Name, func(t *testing.T) {
+			root := expr.RunDSL(t, c.DSL)
+			grpcSvc := root.API.GRPC.Service("Service")
+			require.NotNil(t, grpcSvc)
+			require.Len(t, grpcSvc.GRPCEndpoints, 1)
+			require.Equal(t, c.Expected, grpcSvc.GRPCEndpoints[0].LegacyStreamCompat())
+		})
+	}
+}

expr/testdata/endpoint_dsls.go

@@ -776,3 +776,71 @@ var GRPCEndpointWithStreamingPayloadInitialRequest = func() {
 		})
 	})
 }
+
+var GRPCEndpointStreamCompat = func() {
+	Service("Service", func() {
+		Method("Method", func() {
+			Meta("grpc:stream:compat", "v1")
+			Payload(func() {
+				Field(1, "a", Int)
+				Field(2, "b", ArrayOf(String))
+			})
+			StreamingPayload(Int)
+			GRPC(func() {})
+		})
+	})
+}
+
+var GRPCEndpointStreamCompatServiceLevel = func() {
+	Service("Service", func() {
+		Meta("grpc:stream:compat", "v1")
+		Method("Method", func() {
+			Payload(Int)
+			StreamingPayload(Int)
+			GRPC(func() {})
+		})
+	})
+}
+
+var GRPCEndpointStreamCompatBadValue = func() {
+	Service("Service", func() {
+		Method("Method", func() {
+			Meta("grpc:stream:compat", "v2")
+			Payload(Int)
+			StreamingPayload(Int)
+			GRPC(func() {})
+		})
+	})
+}
+
+var GRPCEndpointStreamCompatNoStreamingPayload = func() {
+	Service("Service", func() {
+		Method("Method", func() {
+			Meta("grpc:stream:compat", "v1")
+			Payload(Int)
+			GRPC(func() {})
+		})
+	})
+}
+
+var GRPCEndpointStreamCompatUnionPayload = func() {
+	var VersionRef = Type("VersionRef", func() {
+		OneOf("ref_type", func() {
+			Field(1, "version_id", String)
+			Field(2, "ref_name", String)
+		})
+		Required("ref_type")
+	})
+	Service("Service", func() {
+		Method("Method", func() {
+			Meta("grpc:stream:compat", "v1")
+			Payload(func() {
+				Field(1, "repository_id", String)
+				Field(2, "version_ref", VersionRef)
+				Required("repository_id", "version_ref")
+			})
+			StreamingPayload(Int)
+			GRPC(func() {})
+		})
+	})
+}

grpc/codegen/client_test.go

@@ -55,6 +55,7 @@ func TestRequestEncoder(t *testing.T) {
 		{"request-encoder-payload-primitive", testdata.ServerStreamingRPCDSL},
 		{"request-encoder-payload-primitive-with-streaming-payload", testdata.ClientStreamingRPCWithPayloadDSL},
 		{"request-encoder-payload-user-type-with-streaming-payload", testdata.BidirectionalStreamingRPCWithPayloadDSL},
+		{"request-encoder-payload-user-type-with-streaming-payload-legacy-compat", testdata.BidirectionalStreamingRPCWithPayloadLegacyCompatDSL},
 		{"request-encoder-payload-with-metadata", testdata.MessageWithMetadataDSL},
 		{"request-encoder-payload-with-validate", testdata.MessageWithValidateDSL},
 		{"request-encoder-payload-with-security-attributes", testdata.MessageWithSecurityAttrsDSL},

grpc/codegen/server.go

@@ -166,7 +166,7 @@ func serverEncodeDecode(genpkg string, svc *expr.GRPCServiceExpr, services *Serv
 				fm["isEmpty"] = isEmpty
 				sections = append(sections, &codegen.SectionTemplate{
 					Name:    "request-decoder",
-					Source:  grpcTemplates.Read(grpcRequestDecoderT, grpcConvertStringToTypeP, "type_conversion", "slice_conversion", "slice_item_conversion"),
+					Source:  grpcTemplates.Read(grpcRequestDecoderT, grpcConvertStringToTypeP, "type_conversion", "slice_conversion", "slice_item_conversion", "metadata_decode"),
 					Data:    e,
 					FuncMap: fm,
 				})

grpc/codegen/server_test.go

@@ -26,6 +26,8 @@ func TestServerGRPCInterface(t *testing.T) {
 		{"bidirectional-streaming-rpc", testdata.BidirectionalStreamingRPCDSL},
 		{"bidirectional-streaming-rpc-with-payload", testdata.BidirectionalStreamingRPCWithPayloadDSL},
 		{"bidirectional-streaming-rpc-with-errors", testdata.BidirectionalStreamingRPCWithErrorsDSL},
+		{"client-streaming-rpc-with-payload-legacy-compat", testdata.ClientStreamingRPCWithPayloadLegacyCompatDSL},
+		{"bidirectional-streaming-rpc-with-payload-legacy-compat", testdata.BidirectionalStreamingRPCWithPayloadLegacyCompatDSL},
 	}
 	for _, c := range cases {
 		t.Run(c.Name, func(t *testing.T) {
@@ -80,6 +82,9 @@ func TestRequestDecoder(t *testing.T) {
 		{"request-decoder-payload-primitive", testdata.ServerStreamingRPCDSL},
 		{"request-decoder-payload-primitive-with-streaming-payload", testdata.ClientStreamingRPCWithPayloadDSL},
 		{"request-decoder-payload-user-type-with-streaming-payload", testdata.BidirectionalStreamingRPCWithPayloadDSL},
+		{"request-decoder-payload-primitive-with-streaming-payload-legacy-compat", testdata.ClientStreamingRPCWithPayloadLegacyCompatDSL},
+		{"request-decoder-payload-user-type-with-streaming-payload-legacy-compat", testdata.BidirectionalStreamingRPCWithPayloadLegacyCompatDSL},
+		{"request-decoder-payload-with-metadata-with-streaming-payload-legacy-compat", testdata.BidirectionalStreamingRPCWithMetadataLegacyCompatDSL},
 		{"request-decoder-payload-with-metadata", testdata.MessageWithMetadataDSL},
 		{"request-decoder-payload-with-validate", testdata.MessageWithValidateDSL},
 		{"request-decoder-payload-with-security-attributes", testdata.MessageWithSecurityAttrsDSL},