client: ignore http status header for gRPC streams

internal/transport/http2_client.go

@@ -1466,10 +1466,11 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 		grpcMessage    string
 		recvCompress   string
 		httpStatusErr  string
-		rawStatusCode  = codes.Internal
+		// the code from the grpc-status header, if present
+		grpcStatusCode = codes.Internal
 		// headerError is set if an error is encountered while parsing the headers
-		headerError        string
-		receivedHTTPStatus string
+		headerError string
+		httpStatus  string
 	)
 
 	for _, hf := range frame.Fields {
@@ -1491,13 +1492,11 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 				t.closeStream(s, se.Err(), true, http2.ErrCodeProtocol, se, nil, endStream)
 				return
 			}
-			rawStatusCode = codes.Code(uint32(code))
+			grpcStatusCode = codes.Code(uint32(code))
 		case "grpc-message":
 			grpcMessage = decodeGrpcMessage(hf.Value)
 		case ":status":
-			if !isGRPC {
-				receivedHTTPStatus = hf.Value
-			}
+			httpStatus = hf.Value
 		default:
 			if isReservedHeader(hf.Name) && !isWhitelistedHeader(hf.Name) {
 				break
@@ -1512,18 +1511,18 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 		}
 	}
 
-	// If a non gRPC response is received, then evaluate entire http status and process close stream / response.
-	// In case http status doesn't provide any error information (status : 200), evalute response code to be Unknown.
+	// If a non gRPC response is received, then evaluate entire http status and
+	//  process close stream / response.
+	// In case http status doesn't provide any error information (status : 200),
+	//  evalute response code to be Unknown.
 	if !isGRPC {
-		var grpcErrorCode = codes.Internal // when header does not include HTTP status, return INTERNAL
-		var errs []string
-
-		switch receivedHTTPStatus {
+		var grpcErrorCode = codes.Internal
+		switch httpStatus {
 		case "":
 			httpStatusErr = "malformed header: missing HTTP status"
 		default:
 			// Any other status code (e.g., "404", "503"). We must parse it.
-			c, err := strconv.ParseInt(receivedHTTPStatus, 10, 32)
+			c, err := strconv.ParseInt(httpStatus, 10, 32)
 			if err != nil {
 				se := status.New(grpcErrorCode, fmt.Sprintf("transport: malformed http-status: %v", err))
 				t.closeStream(s, se.Err(), true, http2.ErrCodeProtocol, se, nil, endStream)
@@ -1536,7 +1535,7 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 						"protocol error: informational header with status code %d must not have END_STREAM set", statusCode))
 					t.closeStream(s, se.Err(), true, http2.ErrCodeProtocol, se, nil, endStream)
 				}
-				//In case of informational headers return
+				// In case of informational headers return
 				return
 			}
 			httpStatusErr = fmt.Sprintf(
@@ -1550,7 +1549,7 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 				grpcErrorCode = codes.Unknown
 			}
 		}
-
+		var errs []string
 		if httpStatusErr != "" {
 			errs = append(errs, httpStatusErr)
 		}
@@ -1613,7 +1612,7 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 		return
 	}
 
-	status := istatus.NewWithProto(rawStatusCode, grpcMessage, mdata[grpcStatusDetailsBinHeader])
+	status := istatus.NewWithProto(grpcStatusCode, grpcMessage, mdata[grpcStatusDetailsBinHeader])
 
 	// If client received END_STREAM from server while stream was still active,
 	// send RST_STREAM.

internal/transport/transport_test.go

@@ -2620,6 +2620,8 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 		wantStatus *status.Status
 		// end stream output
 		wantStatusEndStream *status.Status
+		// head channel closed
+		headerChanClosedForEndStream bool
 	}{
 		{
 			name: "valid header",
@@ -2644,11 +2646,14 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 			},
 			wantStatus: status.New(
 				codes.Unknown,
-				"malformed header: missing HTTP content-type",
+				"unexpected HTTP status code received from server: 200 (OK); malformed header: missing HTTP content-type",
 			),
+			headerChanClosedForEndStream: true,
+			// when headerChan is closed, it is already gRPC and we just need
+			// grpc-status
 			wantStatusEndStream: status.New(
-				codes.Unknown,
-				"malformed header: missing HTTP content-type",
+				codes.OK,
+				"",
 			),
 		},
 		{
@@ -2680,11 +2685,32 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 				codes.Internal,
 				"malformed header: missing HTTP status; transport: received unexpected content-type \"application/json\"",
 			),
+			// This content type will only come when end stream is also initial
+			// header, otherwise we would already be talking gRPC
 			wantStatusEndStream: status.New(
 				codes.Internal,
 				"malformed header: missing HTTP status; transport: received unexpected content-type \"application/json\"",
 			),
 		},
+		{
+			name: "invalid http content type with http status 504 translation",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					{Name: "content-type", Value: "application/json"},
+					{Name: ":status", Value: "504"},
+				},
+			},
+			wantStatus: status.New(
+				codes.Unavailable,
+				"unexpected HTTP status code received from server: 504 (Gateway Timeout); transport: received unexpected content-type \"application/json\"",
+			),
+			// This content type will only come when end stream is also initial
+			// header, otherwise we would already be talking gRPC
+			wantStatusEndStream: status.New(
+				codes.Unavailable,
+				"unexpected HTTP status code received from server: 504 (Gateway Timeout); transport: received unexpected content-type \"application/json\"",
+			),
+		},
 		{
 			name: "http fallback and invalid http status",
 			metaHeaderFrame: &http2.MetaHeadersFrame{
@@ -2697,9 +2723,12 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 				codes.Internal,
 				"transport: malformed http-status: strconv.ParseInt: parsing \"xxxx\": invalid syntax",
 			),
+			// for end stream, when we are already talking gRPC, we expect
+			// grpc - status and fail for it, we will ignore http status.
+			headerChanClosedForEndStream: true,
 			wantStatusEndStream: status.New(
 				codes.Internal,
-				"transport: malformed http-status: strconv.ParseInt: parsing \"xxxx\": invalid syntax",
+				"",
 			),
 		},
 		{
@@ -2747,8 +2776,9 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 					{Name: ":status", Value: "504"},
 				},
 			},
-			wantStatus:          status.New(codes.OK, ""),
-			wantStatusEndStream: status.New(codes.Internal, ""),
+			wantStatus:                   status.New(codes.OK, ""),
+			headerChanClosedForEndStream: true,
+			wantStatusEndStream:          status.New(codes.Internal, ""),
 		},
 		{
 			name: "ignore valid http status for grpc",
@@ -2797,6 +2827,9 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 		})
 		t.Run(fmt.Sprintf("%s-end_stream", test.name), func(t *testing.T) {
 			ts := testStream()
+			if test.headerChanClosedForEndStream {
+				ts.headerChanClosed = 1
+			}
 			s := testClient(ts)
 
 			test.metaHeaderFrame.HeadersFrame = &http2.HeadersFrame{