Repository: grpc/grpc-go

Issue:
ClientConn.Close does not wait for connections to be closed before returning
Hi, I'm testing the compatibility of helloworld example between https://github.com/tower-rs/tower-grpc and https://github.com/grpc/grpc-go.

I found that the Golang client sent a `RST` TCP flag before it closing connection, however, the Rust client sent a `FIN` TCP flag. And, the Rust server think a `RST` flag is not proper.

This issue is born from this https://github.com/tower-rs/tower-grpc/issues/187.

------

### What version of gRPC are you using?
```
const Version = "1.22.0-dev"
```
### What version of Go are you using (`go version`)?
```
$ go version
go version go1.12.6 windows/amd64
```
### What operating system (Linux, Windows, …) and version?
```
Windows 10
```

### What did you do?
If possible, provide a recipe for reproducing the error.

Here is the Wireshark dump of Golang client: https://pastebin.com/XxLkUXQw

And the Screenshot: <img src="https://i.paste.pics/803731625d7ecd1d024abae0e6de8bba.png" width="600" alt="Screenshot">

1. got ConnectionReset  error: https://github.com/tower-rs/tower-grpc/blob/master/tower-grpc-examples/src/helloworld/server.rs + https://github.com/grpc/grpc-go/tree/master/examples/helloworld/greeter_client

2. no this kind error: https://github.com/tower-rs/tower-grpc/blob/master/tower-grpc-examples/src/helloworld/server.rs + https://github.com/tower-rs/tower-grpc/blob/master/tower-grpc-examples/src/helloworld/client.rs

3. got the same kind TCP flag `[RST, ACK]`: https://github.com/grpc/grpc-go/tree/master/examples/helloworld/greeter_server + https://github.com/grpc/grpc-go/tree/master/examples/helloworld/greeter_client

### What did you expect to see?
```
It should be a `FIN` flag instead of `RST` flag when TCP is closing, in my opinion
```

### What did you see instead?
```
A `RST` flag was sent from grpc-go client
```

Discussion:
@menghanl: I believe the root cause is: `ClientConn.Close()` doesn't wait for the underlying goroutines to exit. Which means, when `ClientConn.Close()` returns, the TCP connections are not guaranteed to be closed.

In the helloworld client, `cc.Close()` is the last thing, and the program exits after that. So it's possible that when the program exits, the TCP connection is still open, and will be __force__ closed, resulting in a `RST`.

You can try to add a `defer func() { time.Sleep(time.Second) }()` at the beginning of `main`, to give some for TCP connection to close. You should see `FIN` as expected.

---

The race between `ClientConn.Close()` and other goroutines doesn't seem to be a big problem to me in practice, because it doesn't affect RPCs, and also server should be able to tolerate errors like this.
~~If it still concerns you, please file an issue.~~
EDIT: renaming this issue instead.@dfawley: We still should fix this.  The stale bot went rogue (https://github.com/probot/stale/issues/207) and accidentally tagged this - sorry for the spam.@johnwmstevens: This is still a problem.  Any plans to fix this?@n0izn0iz: bump@dfawley: If anyone would like to help with this, we don't have time to prioritize it, but are willing to review code.@infovivek2020: @purnesh42H Please assign this issue to me
to reproduce this we need to have RUSTC then only we can check right?@purnesh42H: @infovivek2020 I have assigned this to you. You don't need RUSTC. The underlying issue is that client.Close() doesn't wait for all connections to be close. See https://github.com/grpc/grpc-go/issues/2869#issuecomment-503310136@infovivek2020: @purnesh42H on filter the packet in wireshark  for TCP connection after and  before adding delay(#2869) in client main.go I can't see RST Flag 
![image](https://github.com/user-attachments/assets/088009f1-089b-4108-8c63-fd729f0532b7)
getting FIN Flag
@purnesh42H: > #2869

@infovivek2020 could you mention how much delay you tried? Are you not able to repro this with any amount of delay or with no delay at all?@infovivek2020: @purnesh42H tried with hour minute and second  delay but still i am getting FIN signal in wireshark as i have shared previously@infovivek2020: @purnesh42H  panic: cannot update state as grpc.Dial with resolver has not been called

goroutine 1 [running]:
google.golang.org/grpc/resolver/manual.(*Resolver).UpdateState(0xa8d25d0, {{0xa8cede0, 0x3, 0x3}, {0x0, 0x0, 0x0}, 0x0, 0x0})
        /home/vivek/go/src/grpc-go/resolver/manual/manual.go:111 +0x14a
main.main()
        /home/vivek/go/src/grpc-go/examples/features/debugging/client/main.go:69 +0x6e8
exit status 2@purnesh42H: @infovivek2020 was not able to repro this on their machine and asked to work on a different task. Moving it back to unassigned@dfawley: Since this bug is 4 years old, and we've done similar work on similar things in the past, I looked, and I think this is still a problem.  The issue is here:

https://github.com/grpc/grpc-go/blob/47be8a6808cd96f6fb0875c266acf2f3d1981c9c/clientconn.go#L1140-L1149

We shutdown the transports by canceling their contexts (inside `tearDown`) but we never wait for them to be closed.

This is made more difficult because we orphan transports at times, e.g. if we get a GOAWAY on a connection that has a stream open, we create a new connection to the server, but we stop tracking the old one.

Let's ignore the TCP stuff in the issue and focus on the fact that calling `cc.Close` doesn't block until all the goroutines, connections, etc. created by the client have been cleaned up.

Solution patch:
diff --git a/internal/transport/http2_server.go b/internal/transport/http2_server.go
index 11be5599..ffff382e 100644
--- a/internal/transport/http2_server.go
+++ b/internal/transport/http2_server.go
@@ -304,37 +304,92 @@ func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err
 // operateHeader takes action on the decoded headers.
 func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(*Stream), traceCtx func(context.Context, string) context.Context) (fatal bool) {
 	streamID := frame.Header().StreamID
-	state := &decodeState{
-		serverSide: true,
-	}
-	if h2code, err := state.decodeHeader(frame); err != nil {
-		if _, ok := status.FromError(err); ok {
-			t.controlBuf.put(&cleanupStream{
-				streamID: streamID,
-				rst:      true,
-				rstCode:  h2code,
-				onWrite:  func() {},
-			})
-		}
+
+	// frame.Truncated is set to true when framer detects that the current header
+	// list size hits MaxHeaderListSize limit.
+	if frame.Truncated {
+		t.controlBuf.put(&cleanupStream{
+			streamID: streamID,
+			rst:      true,
+			rstCode:  http2.ErrCodeFrameSize,
+			onWrite:  func() {},
+		})
 		return false
 	}
 
 	buf := newRecvBuffer()
 	s := &Stream{
-		id:             streamID,
-		st:             t,
-		buf:            buf,
-		fc:             &inFlow{limit: uint32(t.initialWindowSize)},
-		recvCompress:   state.data.encoding,
-		method:         state.data.method,
-		contentSubtype: state.data.contentSubtype,
+		id:  streamID,
+		st:  t,
+		buf: buf,
+		fc:  &inFlow{limit: uint32(t.initialWindowSize)},
+	}
+
+	var (
+		// If a gRPC Response-Headers has already been received, then it means
+		// that the peer is speaking gRPC and we are in gRPC mode.
+		isGRPC     = false
+		mdata      = make(map[string][]string)
+		httpMethod string
+		// headerError is set if an error is encountered while parsing the headers
+		headerError bool
+
+		timeoutSet bool
+		timeout    time.Duration
+	)
+
+	for _, hf := range frame.Fields {
+		switch hf.Name {
+		case "content-type":
+			contentSubtype, validContentType := grpcutil.ContentSubtype(hf.Value)
+			if !validContentType {
+				break
+			}
+			mdata[hf.Name] = append(mdata[hf.Name], hf.Value)
+			s.contentSubtype = contentSubtype
+			isGRPC = true
+		case "grpc-encoding":
+			s.recvCompress = hf.Value
+		case ":method":
+			httpMethod = hf.Value
+		case ":path":
+			s.method = hf.Value
+		case "grpc-timeout":
+			timeoutSet = true
+			var err error
+			if timeout, err = decodeTimeout(hf.Value); err != nil {
+				headerError = true
+			}
+		default:
+			if isReservedHeader(hf.Name) && !isWhitelistedHeader(hf.Name) {
+				break
+			}
+			v, err := decodeMetadataHeader(hf.Name, hf.Value)
+			if err != nil {
+				headerError = true
+				logger.Warningf("Failed to decode metadata header (%q, %q): %v", hf.Name, hf.Value, err)
+				break
+			}
+			mdata[hf.Name] = append(mdata[hf.Name], v)
+		}
 	}
+
+	if !isGRPC || headerError {
+		t.controlBuf.put(&cleanupStream{
+			streamID: streamID,
+			rst:      true,
+			rstCode:  http2.ErrCodeProtocol,
+			onWrite:  func() {},
+		})
+		return false
+	}
+
 	if frame.StreamEnded() {
 		// s is just created by the caller. No lock needed.
 		s.state = streamReadDone
 	}
-	if state.data.timeoutSet {
-		s.ctx, s.cancel = context.WithTimeout(t.ctx, state.data.timeout)
+	if timeoutSet {
+		s.ctx, s.cancel = context.WithTimeout(t.ctx, timeout)
 	} else {
 		s.ctx, s.cancel = context.WithCancel(t.ctx)
 	}
@@ -347,14 +402,14 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 	}
 	s.ctx = peer.NewContext(s.ctx, pr)
 	// Attach the received metadata to the context.
-	if len(state.data.mdata) > 0 {
-		s.ctx = metadata.NewIncomingContext(s.ctx, state.data.mdata)
-	}
-	if state.data.statsTags != nil {
-		s.ctx = stats.SetIncomingTags(s.ctx, state.data.statsTags)
-	}
-	if state.data.statsTrace != nil {
-		s.ctx = stats.SetIncomingTrace(s.ctx, state.data.statsTrace)
+	if len(mdata) > 0 {
+		s.ctx = metadata.NewIncomingContext(s.ctx, mdata)
+		if statsTags := mdata["grpc-tags-bin"]; len(statsTags) > 0 {
+			s.ctx = stats.SetIncomingTags(s.ctx, []byte(statsTags[len(statsTags)-1]))
+		}
+		if statsTrace := mdata["grpc-trace-bin"]; len(statsTrace) > 0 {
+			s.ctx = stats.SetIncomingTrace(s.ctx, []byte(statsTrace[len(statsTrace)-1]))
+		}
 	}
 	t.mu.Lock()
 	if t.state != reachable {
@@ -383,10 +438,10 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 		return true
 	}
 	t.maxStreamID = streamID
-	if state.data.httpMethod != http.MethodPost {
+	if httpMethod != http.MethodPost {
 		t.mu.Unlock()
 		if logger.V(logLevel) {
-			logger.Warningf("transport: http2Server.operateHeaders parsed a :method field: %v which should be POST", state.data.httpMethod)
+			logger.Infof("transport: http2Server.operateHeaders parsed a :method field: %v which should be POST", httpMethod)
 		}
 		t.controlBuf.put(&cleanupStream{
 			streamID: streamID,
@@ -399,7 +454,7 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 	}
 	if t.inTapHandle != nil {
 		var err error
-		if s.ctx, err = t.inTapHandle(s.ctx, &tap.Info{FullMethodName: state.data.method}); err != nil {
+		if s.ctx, err = t.inTapHandle(s.ctx, &tap.Info{FullMethodName: s.method}); err != nil {
 			t.mu.Unlock()
 			if logger.V(logLevel) {
 				logger.Infof("transport: http2Server.operateHeaders got an error from InTapHandle: %v", err)
@@ -437,7 +492,7 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 			LocalAddr:   t.localAddr,
 			Compression: s.recvCompress,
 			WireLength:  int(frame.Header().Length),
-			Header:      metadata.MD(state.data.mdata).Copy(),
+			Header:      metadata.MD(mdata).Copy(),
 		}
 		t.stats.HandleRPC(s.ctx, inHeader)
 	}
diff --git a/internal/transport/http_util.go b/internal/transport/http_util.go
index 2771e224..15d775fc 100644
--- a/internal/transport/http_util.go
+++ b/internal/transport/http_util.go
@@ -39,7 +39,6 @@ import (
 	spb "google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/grpclog"
-	"google.golang.org/grpc/internal/grpcutil"
 	"google.golang.org/grpc/status"
 )
 
@@ -96,53 +95,6 @@ var (
 	logger = grpclog.Component("transport")
 )
 
-type parsedHeaderData struct {
-	encoding string
-	// statusGen caches the stream status received from the trailer the server
-	// sent.  Client side only.  Do not access directly.  After all trailers are
-	// parsed, use the status method to retrieve the status.
-	statusGen *status.Status
-	// rawStatusCode and rawStatusMsg are set from the raw trailer fields and are not
-	// intended for direct access outside of parsing.
-	rawStatusCode *int
-	rawStatusMsg  string
-	httpStatus    *int
-	// Server side only fields.
-	timeoutSet bool
-	timeout    time.Duration
-	method     string
-	httpMethod string
-	// key-value metadata map from the peer.
-	mdata          map[string][]string
-	statsTags      []byte
-	statsTrace     []byte
-	contentSubtype string
-
-	// isGRPC field indicates whether the peer is speaking gRPC (otherwise HTTP).
-	//
-	// We are in gRPC mode (peer speaking gRPC) if:
-	// 	* We are client side and have already received a HEADER frame that indicates gRPC peer.
-	//  * The header contains valid  a content-type, i.e. a string starts with "application/grpc"
-	// And we should handle error specific to gRPC.
-	//
-	// Otherwise (i.e. a content-type string starts without "application/grpc", or does not exist), we
-	// are in HTTP fallback mode, and should handle error specific to HTTP.
-	isGRPC         bool
-	grpcErr        error
-	httpErr        error
-	contentTypeErr string
-}
-
-// decodeState configures decoding criteria and records the decoded data.
-type decodeState struct {
-	// whether decoding on server side or not
-	serverSide bool
-
-	// Records the states during HPACK decoding. It will be filled with info parsed from HTTP HEADERS
-	// frame once decodeHeader function has been invoked and returned.
-	data parsedHeaderData
-}
-
 // isReservedHeader checks whether hdr belongs to HTTP2 headers
 // reserved by gRPC protocol. Any other headers are classified as the
 // user-specified metadata.
@@ -221,63 +173,6 @@ func decodeGRPCStatusDetails(rawDetails string) (*status.Status, error) {
 	return status.FromProto(st), nil
 }
 
-func (d *decodeState) decodeHeader(frame *http2.MetaHeadersFrame) (http2.ErrCode, error) {
-	// frame.Truncated is set to true when framer detects that the current header
-	// list size hits MaxHeaderListSize limit.
-	if frame.Truncated {
-		return http2.ErrCodeFrameSize, status.Error(codes.Internal, "peer header list size exceeded limit")
-	}
-
-	for _, hf := range frame.Fields {
-		d.processHeaderField(hf)
-	}
-
-	if d.data.isGRPC {
-		if d.data.grpcErr != nil {
-			return http2.ErrCodeProtocol, d.data.grpcErr
-		}
-		if d.serverSide {
-			return http2.ErrCodeNo, nil
-		}
-		if d.data.rawStatusCode == nil && d.data.statusGen == nil {
-			// gRPC status doesn't exist.
-			// Set rawStatusCode to be unknown and return nil error.
-			// So that, if the stream has ended this Unknown status
-			// will be propagated to the user.
-			// Otherwise, it will be ignored. In which case, status from
-			// a later trailer, that has StreamEnded flag set, is propagated.
-			code := int(codes.Unknown)
-			d.data.rawStatusCode = &code
-		}
-		return http2.ErrCodeNo, nil
-	}
-
-	// HTTP fallback mode
-	if d.data.httpErr != nil {
-		return http2.ErrCodeProtocol, d.data.httpErr
-	}
-
-	var (
-		code = codes.Internal // when header does not include HTTP status, return INTERNAL
-		ok   bool
-	)
-
-	if d.data.httpStatus != nil {
-		code, ok = HTTPStatusConvTab[*(d.data.httpStatus)]
-		if !ok {
-			code = codes.Unknown
-		}
-	}
-
-	return http2.ErrCodeProtocol, status.Error(code, d.constructHTTPErrMsg())
-}
-
-// constructErrMsg constructs error message to be returned in HTTP fallback mode.
-// Format: HTTP status code and its corresponding message + content-type error message.
-func (d *decodeState) constructHTTPErrMsg() string {
-	return constructHTTPErrMsg(d.data.httpStatus, d.data.contentTypeErr)
-}
-
 // constructErrMsg constructs error message to be returned in HTTP fallback mode.
 // Format: HTTP status code and its corresponding message + content-type error message.
 func constructHTTPErrMsg(httpStatus *int, contentTypeErr string) string {
@@ -298,99 +193,6 @@ func constructHTTPErrMsg(httpStatus *int, contentTypeErr string) string {
 	return strings.Join(errMsgs, "; ")
 }
 
-func (d *decodeState) addMetadata(k, v string) {
-	if d.data.mdata == nil {
-		d.data.mdata = make(map[string][]string)
-	}
-	d.data.mdata[k] = append(d.data.mdata[k], v)
-}
-
-func (d *decodeState) processHeaderField(f hpack.HeaderField) {
-	switch f.Name {
-	case "content-type":
-		contentSubtype, validContentType := grpcutil.ContentSubtype(f.Value)
-		if !validContentType {
-			d.data.contentTypeErr = fmt.Sprintf("transport: received the unexpected content-type %q", f.Value)
-			return
-		}
-		d.data.contentSubtype = contentSubtype
-		// TODO: do we want to propagate the whole content-type in the metadata,
-		// or come up with a way to just propagate the content-subtype if it was set?
-		// ie {"content-type": "application/grpc+proto"} or {"content-subtype": "proto"}
-		// in the metadata?
-		d.addMetadata(f.Name, f.Value)
-		d.data.isGRPC = true
-	case "grpc-encoding":
-		d.data.encoding = f.Value
-	case "grpc-status":
-		code, err := strconv.Atoi(f.Value)
-		if err != nil {
-			d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-status: %v", err)
-			return
-		}
-		d.data.rawStatusCode = &code
-	case "grpc-message":
-		d.data.rawStatusMsg = decodeGrpcMessage(f.Value)
-	case "grpc-status-details-bin":
-		v, err := decodeBinHeader(f.Value)
-		if err != nil {
-			d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
-			return
-		}
-		s := &spb.Status{}
-		if err := proto.Unmarshal(v, s); err != nil {
-			d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
-			return
-		}
-		d.data.statusGen = status.FromProto(s)
-	case "grpc-timeout":
-		d.data.timeoutSet = true
-		var err error
-		if d.data.timeout, err = decodeTimeout(f.Value); err != nil {
-			d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed time-out: %v", err)
-		}
-	case ":path":
-		d.data.method = f.Value
-	case ":status":
-		code, err := strconv.Atoi(f.Value)
-		if err != nil {
-			d.data.httpErr = status.Errorf(codes.Internal, "transport: malformed http-status: %v", err)
-			return
-		}
-		d.data.httpStatus = &code
-	case "grpc-tags-bin":
-		v, err := decodeBinHeader(f.Value)
-		if err != nil {
-			d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-tags-bin: %v", err)
-			return
-		}
-		d.data.statsTags = v
-		d.addMetadata(f.Name, string(v))
-	case "grpc-trace-bin":
-		v, err := decodeBinHeader(f.Value)
-		if err != nil {
-			d.data.grpcErr = status.Errorf(codes.Internal, "transport: malformed grpc-trace-bin: %v", err)
-			return
-		}
-		d.data.statsTrace = v
-		d.addMetadata(f.Name, string(v))
-	case ":method":
-		d.data.httpMethod = f.Value
-	default:
-		if isReservedHeader(f.Name) && !isWhitelistedHeader(f.Name) {
-			break
-		}
-		v, err := decodeMetadataHeader(f.Name, f.Value)
-		if err != nil {
-			if logger.V(logLevel) {
-				logger.Errorf("Failed to decode metadata header (%q, %q): %v", f.Name, f.Value, err)
-			}
-			return
-		}
-		d.addMetadata(f.Name, v)
-	}
-}
-
 type timeoutUnit uint8
 
 const (
diff --git a/interop/xds/client/Dockerfile b/interop/xds/client/Dockerfile
new file mode 100644
index 00000000..060f8a8c
--- /dev/null
+++ b/interop/xds/client/Dockerfile
@@ -0,0 +1,34 @@
+# Copyright 2021 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Dockerfile for building the xDS interop client. To build the image, run the
+# following command from grpc-go directory:
+# docker build -t <TAG> -f interop/xds/client/Dockerfile .
+
+FROM golang:1.16-alpine as build
+
+# Make a grpc-go directory and copy the repo into it.
+WORKDIR /go/src/grpc-go
+COPY . .
+
+# Build a static binary without cgo so that we can copy just the binary in the
+# final image, and can get rid of Go compiler and gRPC-Go dependencies.
+RUN go build -tags osusergo,netgo interop/xds/client/client.go
+
+# Second stage of the build which copies over only the client binary and skips
+# the Go compiler and gRPC repo from the earlier stage. This significantly
+# reduces the docker image size.
+FROM alpine
+COPY --from=build /go/src/grpc-go/client .
+ENTRYPOINT ["./client"]



Test patch:
diff --git a/internal/transport/http_util_test.go b/internal/transport/http_util_test.go
index 2205050a..bbd53180 100644
--- a/internal/transport/http_util_test.go
+++ b/internal/transport/http_util_test.go
@@ -23,9 +23,6 @@ import (
 	"reflect"
 	"testing"
 	"time"
-
-	"golang.org/x/net/http2"
-	"golang.org/x/net/http2/hpack"
 )
 
 func (s) TestTimeoutDecode(t *testing.T) {
@@ -189,68 +186,6 @@ func (s) TestDecodeMetadataHeader(t *testing.T) {
 	}
 }
 
-func (s) TestDecodeHeaderH2ErrCode(t *testing.T) {
-	for _, test := range []struct {
-		name string
-		// input
-		metaHeaderFrame *http2.MetaHeadersFrame
-		serverSide      bool
-		// output
-		wantCode http2.ErrCode
-	}{
-		{
-			name: "valid header",
-			metaHeaderFrame: &http2.MetaHeadersFrame{Fields: []hpack.HeaderField{
-				{Name: "content-type", Value: "application/grpc"},
-			}},
-			wantCode: http2.ErrCodeNo,
-		},
-		{
-			name: "valid header serverSide",
-			metaHeaderFrame: &http2.MetaHeadersFrame{Fields: []hpack.HeaderField{
-				{Name: "content-type", Value: "application/grpc"},
-			}},
-			serverSide: true,
-			wantCode:   http2.ErrCodeNo,
-		},
-		{
-			name: "invalid grpc status header field",
-			metaHeaderFrame: &http2.MetaHeadersFrame{Fields: []hpack.HeaderField{
-				{Name: "content-type", Value: "application/grpc"},
-				{Name: "grpc-status", Value: "xxxx"},
-			}},
-			wantCode: http2.ErrCodeProtocol,
-		},
-		{
-			name: "invalid http content type",
-			metaHeaderFrame: &http2.MetaHeadersFrame{Fields: []hpack.HeaderField{
-				{Name: "content-type", Value: "application/json"},
-			}},
-			wantCode: http2.ErrCodeProtocol,
-		},
-		{
-			name: "http fallback and invalid http status",
-			metaHeaderFrame: &http2.MetaHeadersFrame{Fields: []hpack.HeaderField{
-				// No content type provided then fallback into handling http error.
-				{Name: ":status", Value: "xxxx"},
-			}},
-			wantCode: http2.ErrCodeProtocol,
-		},
-		{
-			name:            "http2 frame size exceeds",
-			metaHeaderFrame: &http2.MetaHeadersFrame{Fields: nil, Truncated: true},
-			wantCode:        http2.ErrCodeFrameSize,
-		},
-	} {
-		t.Run(test.name, func(t *testing.T) {
-			state := &decodeState{serverSide: test.serverSide}
-			if h2code, _ := state.decodeHeader(test.metaHeaderFrame); h2code != test.wantCode {
-				t.Fatalf("decodeState.decodeHeader(%v) = %v, want %v", test.metaHeaderFrame, h2code, test.wantCode)
-			}
-		})
-	}
-}
-
 func (s) TestParseDialTarget(t *testing.T) {
 	for _, test := range []struct {
 		target, wantNet, wantAddr string
diff --git a/internal/transport/transport_test.go b/internal/transport/transport_test.go
index c3830a8f..3aef8627 100644
--- a/internal/transport/transport_test.go
+++ b/internal/transport/transport_test.go
@@ -1976,3 +1976,110 @@ func (s) TestClientHandshakeInfo(t *testing.T) {
 		t.Fatalf("received attributes %v in creds, want %v", gotAttr, wantAttr)
 	}
 }
+
+func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
+	for _, test := range []struct {
+		name string
+		// input
+		metaHeaderFrame *http2.MetaHeadersFrame
+		// output
+		wantStatus *status.Status
+	}{
+		{
+			name: "valid header",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					{Name: "content-type", Value: "application/grpc"},
+					{Name: "grpc-status", Value: "0"},
+				},
+			},
+			// no error
+			wantStatus: status.New(codes.OK, ""),
+		},
+		{
+			name: "invalid grpc status header field",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					{Name: "content-type", Value: "application/grpc"},
+					{Name: "grpc-status", Value: "xxxx"},
+				},
+			},
+			wantStatus: status.New(
+				codes.Internal,
+				"transport: malformed grpc-status: strconv.ParseInt: parsing \"xxxx\": invalid syntax",
+			),
+		},
+		{
+			name: "invalid http content type",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					{Name: "content-type", Value: "application/json"},
+				},
+			},
+			wantStatus: status.New(
+				codes.Internal,
+				": HTTP status code 0; transport: received the unexpected content-type \"application/json\"",
+			),
+		},
+		{
+			name: "http fallback and invalid http status",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					// No content type provided then fallback into handling http error.
+					{Name: ":status", Value: "xxxx"},
+				},
+			},
+			wantStatus: status.New(
+				codes.Internal,
+				"transport: malformed http-status: strconv.ParseInt: parsing \"xxxx\": invalid syntax",
+			),
+		},
+		{
+			name: "http2 frame size exceeds",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields:    nil,
+				Truncated: true,
+			},
+			wantStatus: status.New(
+				codes.Internal,
+				"peer header list size exceeded limit",
+			),
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			ts := &Stream{
+				done:       make(chan struct{}),
+				headerChan: make(chan struct{}),
+				buf: &recvBuffer{
+					c:  make(chan recvMsg),
+					mu: sync.Mutex{},
+				},
+			}
+			s := &http2Client{
+				mu: sync.Mutex{},
+				activeStreams: map[uint32]*Stream{
+					0: ts,
+				},
+				controlBuf: &controlBuffer{
+					ch:   make(chan struct{}),
+					done: make(chan struct{}),
+					list: &itemList{},
+				},
+			}
+			test.metaHeaderFrame.HeadersFrame = &http2.HeadersFrame{
+				FrameHeader: http2.FrameHeader{
+					StreamID: 0,
+					Flags:    http2.FlagHeadersEndStream,
+				},
+			}
+
+			s.operateHeaders(test.metaHeaderFrame)
+
+			got := ts.status
+			want := test.wantStatus
+			if got.Code() != want.Code() || got.Message() != want.Message() {
+				t.Fatalf("operateHeaders(%v); status = %v; want %v", test.metaHeaderFrame, got, want)
+			}
+		})
+	}
+}