Repository: grpc/grpc-go

Issue:
Flaky test: leakcheck.TestCheck
It seems there may have been a change in the stdlib that makes `WithTimeout` spawn a goroutine which may still be alive when we run `CheckGoroutines` in this test.

```
--- FAIL: TestCheck (2.05s)
    leakcheck_test.go:58: CheckGoroutines found 4 leaks, want 3 leaks
    leakcheck_test.go:59: leaked goroutines:
<snip out expected goroutines>
        Leaked goroutine: goroutine 8 [runnable]:
        context.WithDeadlineCause.func2()
        	/opt/hostedtoolcache/go/1.24.2/x64/src/context/context.go:643
        created by time.goFunc
        	/opt/hostedtoolcache/go/1.24.2/x64/src/time/sleep.go:215 +0x38
FAIL
FAIL	google.golang.org/grpc/internal/leakcheck	12.308s
```

We can probably just add it to https://github.com/grpc/grpc-go/blob/4cedec40eb2ccfbe3f56bb15e894903111ada2d2/internal/leakcheck/leakcheck.go#L172

We should also tidy up these error messages here while we're here:

https://github.com/grpc/grpc-go/blob/4cedec40eb2ccfbe3f56bb15e894903111ada2d2/internal/leakcheck/leakcheck_test.go#L51
https://github.com/grpc/grpc-go/blob/4cedec40eb2ccfbe3f56bb15e894903111ada2d2/internal/leakcheck/leakcheck_test.go#L78


EDIT: link to failure: https://github.com/grpc/grpc-go/actions/runs/14715401045/job/41297489567?pr=8155#step:8:116

Discussion:
@arjan-bal: @Pranjali-2501 is working on this.@arjan-bal: The root cause is that a context with a deadline [calls  time.AfterFunc to execute a function on deadline expiry](https://github.com/golang/go/blob/34c8b14ca9f4096383d658fbd748322a993a2bd2/src/context/context.go#L643-L646). This results in a goroutine being spawned only when the context expires. We check for goroutines in a loop here until the context expires: https://github.com/grpc/grpc-go/blob/4cedec40eb2ccfbe3f56bb15e894903111ada2d2/internal/leakcheck/leakcheck.go#L252-L257

Assume the following sequence of events when the context is very close to expiring:
1. ctx.Err() == nil, so we enter the loop.
2. timer created by the context fires, spawning a goroutine.
3. In the for loop, we capture the running goroutines by calling `leaked = interestingGoroutines()`. The goroutine created by the timer is captured.
4. goroutine spawned by the timer finishes, setting ctx.Err().
5. In the next iteration of the loop `ctx.Err() != nil` so the loop breaks.

This race is infrequent because it happens when the context expires after we enter the for loop but before we capture the running goroutines.

One solution would be to not update the `leaked` variable immediately. Instead, store the running goroutines in a temp variable, check if `ctx.Err()` is still nil after capturing the running goroutines, if yes, then update `leaked`.@dfawley: One thing I just thought of that can be changed in this test: let's cause the goroutines to leak by having them block on a channel instead of sleeping.  So we run `interestingGoroutines` and expect to find them, then we close the channel and call `CheckGoroutines`, and expect it to succeed.

For the other (main) issue, I think we can just ensure `interestingGoroutines` retruns `3` and then expect `CheckGoroutines` to fail without caring about exactly how it failed.@Pranjali-2501: I have reproduced the error mentioned in this issue by introducing an additional delay that triggers after the timeout mentioned in [context.WithTimeout(context.Background(), time.Second)](https://github.com/grpc/grpc-go/blob/b3d63b180cdf8597ac95f61a697ba08f6161594b/internal/leakcheck/leakcheck_test.go#L54) has expired, but before it calls f func() in its own goroutine. 

```
=== RUN   TestCheck
    leakcheck_test.go:67: CheckGoroutines found 4 leaks, want 3 leaks
    leakcheck_test.go:68: leaked goroutines:
       <expected goutines>
        Leaked goroutine: goroutine 82 [sleep]:
        time.Sleep(0x5f5e100)
                /usr/local/google/home/pranjalisaxena/.asdf/installs/golang/1.24.2/go/src/runtime/time.go:338 +0x165
        context.WithDeadlineCause.func2()
                /usr/local/google/home/pranjalisaxena/.asdf/installs/golang/1.24.2/go/src/context/context.go:645 +0x33
        created by time.goFunc
                /usr/local/google/home/pranjalisaxena/.asdf/installs/golang/1.24.2/go/src/time/sleep.go:215 +0x2d
        
      --- FAIL: TestCheck (2.05s)
  FAIL
  exit status 1
  FAIL    google.golang.org/grpc/internal/leakcheck       2.055s
```

I'm able to solve it by adding a string `WithDeadlineCause` in  [ var goroutinesToIgnore = []string{ ](https://github.com/grpc/grpc-go/blob/b3d63b180cdf8597ac95f61a697ba08f6161594b/internal/leakcheck/leakcheck.go#L172C1-L192C2).

@dfawley and @arjan-bal , please let me know if we need to implement a general solution to handle all potential failure scenarios, or if addressing this specific failure is sufficient.@dfawley: I don't think it's a good idea to add things to the ignore list unless they are going to be sticking around in the system in a way we can't control.  To fix this test so it isn't flaky, it should be sufficient to do what I suggested above: change the test to confirm that `interestingGoroutines() == 3` and then ensure that `Check` fails without checking how many things showed up in the error.

Solution patch:
diff --git a/Documentation/encoding.md b/Documentation/encoding.md
index cce9384f..00e19cc8 100644
--- a/Documentation/encoding.md
+++ b/Documentation/encoding.md
@@ -57,7 +57,7 @@ As a reminder, all `CallOption`s may be converted into `DialOption`s that become
 the default for all RPCs sent through a client using `grpc.WithDefaultCallOptions`:
 
 ```go
-    myclient := grpc.Dial(ctx, target, grpc.WithDefaultCallOptions(grpc.CallContentSubtype("mycodec")))
+    myclient := grpc.NewClient(target, grpc.WithDefaultCallOptions(grpc.CallContentSubtype("mycodec")))
 ```
 
 When specified in either of these ways, messages will be encoded using this
@@ -132,7 +132,7 @@ As a reminder, all `CallOption`s may be converted into `DialOption`s that become
 the default for all RPCs sent through a client using `grpc.WithDefaultCallOptions`:
 
 ```go
-    myclient := grpc.Dial(ctx, target, grpc.WithDefaultCallOptions(grpc.UseCompressor("gzip")))
+    myclient := grpc.NewClient(target, grpc.WithDefaultCallOptions(grpc.UseCompressor("gzip")))
 ```
 
 When specified in either of these ways, messages will be compressed using this
diff --git a/Documentation/grpc-auth-support.md b/Documentation/grpc-auth-support.md
index cedcaa6f..c8c2a556 100644
--- a/Documentation/grpc-auth-support.md
+++ b/Documentation/grpc-auth-support.md
@@ -5,7 +5,7 @@ As outlined in the [gRPC authentication guide](https://grpc.io/docs/guides/auth.
 # Enabling TLS on a gRPC client
 
 ```Go
-conn, err := grpc.Dial(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")))
+conn, err := grpc.NewClient(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")))
 ```
 
 # Enabling TLS on a gRPC server
@@ -63,7 +63,7 @@ to prevent any insecure transmission of tokens.
 ## Google Compute Engine (GCE)
 
 ```Go
-conn, err := grpc.Dial(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(oauth.NewComputeEngine()))
+conn, err := grpc.NewClient(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(oauth.NewComputeEngine()))
 ```
 
 ## JWT
@@ -73,6 +73,6 @@ jwtCreds, err := oauth.NewServiceAccountFromFile(*serviceAccountKeyFile, *oauthS
 if err != nil {
   log.Fatalf("Failed to create JWT credentials: %v", err)
 }
-conn, err := grpc.Dial(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(jwtCreds))
+conn, err := grpc.NewClient(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(jwtCreds))
 ```
 
diff --git a/credentials/insecure/insecure.go b/credentials/insecure/insecure.go
index f45c04f7..93156c0f 100644
--- a/credentials/insecure/insecure.go
+++ b/credentials/insecure/insecure.go
@@ -30,7 +30,7 @@ import (
 // NewCredentials returns a credentials which disables transport security.
 //
 // Note that using this credentials with per-RPC credentials which require
-// transport security is incompatible and will cause grpc.Dial() to fail.
+// transport security is incompatible and will cause RPCs to fail.
 func NewCredentials() credentials.TransportCredentials {
 	return insecureTC{}
 }
diff --git a/dialoptions.go b/dialoptions.go
index 405a2ffe..050ba0f1 100644
--- a/dialoptions.go
+++ b/dialoptions.go
@@ -360,7 +360,7 @@ func WithReturnConnectionError() DialOption {
 //
 // Note that using this DialOption with per-RPC credentials (through
 // WithCredentialsBundle or WithPerRPCCredentials) which require transport
-// security is incompatible and will cause grpc.Dial() to fail.
+// security is incompatible and will cause RPCs to fail.
 //
 // Deprecated: use WithTransportCredentials and insecure.NewCredentials()
 // instead. Will be supported throughout 1.x.
diff --git a/internal/grpcsync/event.go b/internal/grpcsync/event.go
index fbe697c3..d788c249 100644
--- a/internal/grpcsync/event.go
+++ b/internal/grpcsync/event.go
@@ -21,28 +21,25 @@
 package grpcsync
 
 import (
-	"sync"
 	"sync/atomic"
 )
 
 // Event represents a one-time event that may occur in the future.
 type Event struct {
-	fired int32
+	fired atomic.Bool
 	c     chan struct{}
-	o     sync.Once
 }
 
 // Fire causes e to complete.  It is safe to call multiple times, and
 // concurrently.  It returns true iff this call to Fire caused the signaling
-// channel returned by Done to close.
+// channel returned by Done to close. If Fire returns false, it is possible
+// the Done channel has not been closed yet.
 func (e *Event) Fire() bool {
-	ret := false
-	e.o.Do(func() {
-		atomic.StoreInt32(&e.fired, 1)
+	if e.fired.CompareAndSwap(false, true) {
 		close(e.c)
-		ret = true
-	})
-	return ret
+		return true
+	}
+	return false
 }
 
 // Done returns a channel that will be closed when Fire is called.
@@ -52,7 +49,7 @@ func (e *Event) Done() <-chan struct{} {
 
 // HasFired returns true if Fire has been called.
 func (e *Event) HasFired() bool {
-	return atomic.LoadInt32(&e.fired) == 1
+	return e.fired.Load()
 }
 
 // NewEvent returns a new, ready-to-use Event.
diff --git a/internal/transport/http2_client.go b/internal/transport/http2_client.go
index ac0fd4b3..ed2d93af 100644
--- a/internal/transport/http2_client.go
+++ b/internal/transport/http2_client.go
@@ -1261,7 +1261,8 @@ func (t *http2Client) handleRSTStream(f *http2.RSTStreamFrame) {
 			statusCode = codes.DeadlineExceeded
 		}
 	}
-	t.closeStream(s, io.EOF, false, http2.ErrCodeNo, status.Newf(statusCode, "stream terminated by RST_STREAM with error code: %v", f.ErrCode), nil, false)
+	st := status.Newf(statusCode, "stream terminated by RST_STREAM with error code: %v", f.ErrCode)
+	t.closeStream(s, st.Err(), false, http2.ErrCodeNo, st, nil, false)
 }
 
 func (t *http2Client) handleSettings(f *http2.SettingsFrame, isFirst bool) {
diff --git a/internal/transport/http_util.go b/internal/transport/http_util.go
index f997f9fd..607d2c4c 100644
--- a/internal/transport/http_util.go
+++ b/internal/transport/http_util.go
@@ -196,11 +196,14 @@ func decodeTimeout(s string) (time.Duration, error) {
 	if !ok {
 		return 0, fmt.Errorf("transport: timeout unit is not recognized: %q", s)
 	}
-	t, err := strconv.ParseInt(s[:size-1], 10, 64)
+	t, err := strconv.ParseUint(s[:size-1], 10, 64)
 	if err != nil {
 		return 0, err
 	}
-	const maxHours = math.MaxInt64 / int64(time.Hour)
+	if t == 0 {
+		return 0, fmt.Errorf("transport: timeout must be positive: %q", s)
+	}
+	const maxHours = math.MaxInt64 / uint64(time.Hour)
 	if d == time.Hour && t > maxHours {
 		// This timeout would overflow math.MaxInt64; clamp it.
 		return time.Duration(math.MaxInt64), nil



Test patch:
diff --git a/balancer/endpointsharding/endpointsharding_test.go b/balancer/endpointsharding/endpointsharding_test.go
index ebad977e..5567b4c9 100644
--- a/balancer/endpointsharding/endpointsharding_test.go
+++ b/balancer/endpointsharding/endpointsharding_test.go
@@ -23,7 +23,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"log"
 	"strings"
 	"testing"
 	"time"
@@ -166,7 +165,7 @@ func (s) TestEndpointShardingBasic(t *testing.T) {
 	}
 	cc, err := grpc.NewClient(mr.Scheme()+":///", dOpts...)
 	if err != nil {
-		log.Fatalf("Failed to create new client: %v", err)
+		t.Fatalf("Failed to create new client: %v", err)
 	}
 	defer cc.Close()
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
@@ -246,7 +245,7 @@ func (s) TestEndpointShardingReconnectDisabled(t *testing.T) {
 
 	cc, err := grpc.NewClient(mr.Scheme()+":///", grpc.WithResolvers(mr), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
-		log.Fatalf("Failed to create new client: %v", err)
+		t.Fatalf("Failed to create new client: %v", err)
 	}
 	defer cc.Close()
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
diff --git a/internal/leakcheck/leakcheck_test.go b/internal/leakcheck/leakcheck_test.go
index 26ecfc14..3de3a1b4 100644
--- a/internal/leakcheck/leakcheck_test.go
+++ b/internal/leakcheck/leakcheck_test.go
@@ -44,20 +44,21 @@ func (e *testLogger) Errorf(format string, args ...any) {
 
 func TestCheck(t *testing.T) {
 	const leakCount = 3
+	ch := make(chan struct{})
 	for i := 0; i < leakCount; i++ {
-		go func() { time.Sleep(2 * time.Second) }()
+		go func() { <-ch }()
 	}
-	if ig := interestingGoroutines(); len(ig) == 0 {
-		t.Error("blah")
+	if leaked := interestingGoroutines(); len(leaked) != leakCount {
+		t.Errorf("interestingGoroutines() = %d, want length %d", len(leaked), leakCount)
 	}
 	e := &testLogger{}
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	CheckGoroutines(ctx, e)
-	if e.errorCount != leakCount {
-		t.Errorf("CheckGoroutines found %v leaks, want %v leaks", e.errorCount, leakCount)
+	if CheckGoroutines(ctx, e); e.errorCount < leakCount {
+		t.Errorf("CheckGoroutines() = %d, want count %d", e.errorCount, leakCount)
 		t.Logf("leaked goroutines:\n%v", strings.Join(e.errors, "\n"))
 	}
+	close(ch)
 	ctx, cancel = context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()
 	CheckGoroutines(ctx, t)
@@ -69,22 +70,23 @@ func ignoredTestingLeak(d time.Duration) {
 
 func TestCheckRegisterIgnore(t *testing.T) {
 	RegisterIgnoreGoroutine("ignoredTestingLeak")
+	go ignoredTestingLeak(3 * time.Second)
 	const leakCount = 3
+	ch := make(chan struct{})
 	for i := 0; i < leakCount; i++ {
-		go func() { time.Sleep(2 * time.Second) }()
+		go func() { <-ch }()
 	}
-	go func() { ignoredTestingLeak(3 * time.Second) }()
-	if ig := interestingGoroutines(); len(ig) == 0 {
-		t.Error("blah")
+	if leaked := interestingGoroutines(); len(leaked) != leakCount {
+		t.Errorf("interestingGoroutines() = %d, want length %d", len(leaked), leakCount)
 	}
 	e := &testLogger{}
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	CheckGoroutines(ctx, e)
-	if e.errorCount != leakCount {
-		t.Errorf("CheckGoroutines found %v leaks, want %v leaks", e.errorCount, leakCount)
+	if CheckGoroutines(ctx, e); e.errorCount < leakCount {
+		t.Errorf("CheckGoroutines() = %d, want count %d", e.errorCount, leakCount)
 		t.Logf("leaked goroutines:\n%v", strings.Join(e.errors, "\n"))
 	}
+	close(ch)
 	ctx, cancel = context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()
 	CheckGoroutines(ctx, t)
diff --git a/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go b/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go
index 602c5372..b7176d5f 100644
--- a/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go
+++ b/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go
@@ -131,15 +131,29 @@ func (s) TestDelegatingResolverNoProxyEnvVarsSet(t *testing.T) {
 // overwriting the previously registered DNS resolver. This allows the test to
 // mock the DNS resolution for the proxy resolver. It also registers the
 // original DNS resolver after the test is done.
-func setupDNS(t *testing.T) *manual.Resolver {
+func setupDNS(t *testing.T) (*manual.Resolver, chan struct{}) {
 	t.Helper()
 	mr := manual.NewBuilderWithScheme("dns")
 
 	dnsResolverBuilder := resolver.Get("dns")
 	resolver.Register(mr)
 
+	resolverBuilt := make(chan struct{})
+	mr.BuildCallback = func(resolver.Target, resolver.ClientConn, resolver.BuildOptions) {
+		close(resolverBuilt)
+	}
+
 	t.Cleanup(func() { resolver.Register(dnsResolverBuilder) })
-	return mr
+	return mr, resolverBuilt
+}
+
+func mustBuildResolver(ctx context.Context, t *testing.T, buildCh chan struct{}) {
+	t.Helper()
+	select {
+	case <-buildCh:
+	case <-ctx.Done():
+		t.Fatalf("Context timed out waiting for resolver to be built.")
+	}
 }
 
 // proxyAddressWithTargetAttribute creates a resolver.Address for the proxy,
@@ -181,7 +195,7 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithTargetResolution(t *testing.T)
 	targetResolver := manual.NewBuilderWithScheme("dns")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, true); err != nil {
@@ -202,6 +216,11 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithTargetResolution(t *testing.T)
 	case <-time.After(defaultTestShortTimeout):
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
 	proxyResolver.UpdateState(resolver.State{
 		Addresses:     []resolver.Address{{Addr: resolvedProxyTestAddr1}},
 		ServiceConfig: &serviceconfig.ParseResult{},
@@ -218,8 +237,8 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithTargetResolution(t *testing.T)
 	var gotState resolver.State
 	select {
 	case gotState = <-stateCh:
-	case <-time.After(defaultTestTimeout):
-		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	case <-ctx.Done():
+		t.Fatal("Context timeed out when waiting for a state update from the delegating resolver")
 	}
 
 	if diff := cmp.Diff(gotState, wantState); diff != "" {
@@ -257,13 +276,18 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithNoTargetResolution(t *testing.
 	targetResolver := manual.NewBuilderWithScheme("dns")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
 	proxyResolver.UpdateState(resolver.State{
 		Addresses: []resolver.Address{
 			{Addr: resolvedProxyTestAddr1},
@@ -278,8 +302,8 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithNoTargetResolution(t *testing.
 	var gotState resolver.State
 	select {
 	case gotState = <-stateCh:
-	case <-time.After(defaultTestTimeout):
-		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	case <-ctx.Done():
+		t.Fatal("Context timed out when waiting for a state update from the delegating resolver")
 	}
 
 	if diff := cmp.Diff(gotState, wantState); diff != "" {
@@ -319,7 +343,7 @@ func (s) TestDelegatingResolverwithCustomResolverAndProxy(t *testing.T) {
 	targetResolver := manual.NewBuilderWithScheme("test")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
@@ -340,6 +364,11 @@ func (s) TestDelegatingResolverwithCustomResolverAndProxy(t *testing.T) {
 	case <-time.After(defaultTestShortTimeout):
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
 	proxyResolver.UpdateState(resolver.State{
 		Addresses:     []resolver.Address{{Addr: resolvedProxyTestAddr1}},
 		ServiceConfig: &serviceconfig.ParseResult{},
@@ -355,8 +384,8 @@ func (s) TestDelegatingResolverwithCustomResolverAndProxy(t *testing.T) {
 	var gotState resolver.State
 	select {
 	case gotState = <-stateCh:
-	case <-time.After(defaultTestTimeout):
-		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	case <-ctx.Done():
+		t.Fatal("Context timed out when waiting for a state update from the delegating resolver")
 	}
 
 	if diff := cmp.Diff(gotState, wantState); diff != "" {
@@ -401,7 +430,7 @@ func (s) TestDelegatingResolverForEndpointsWithProxy(t *testing.T) {
 	targetResolver := manual.NewBuilderWithScheme("test")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
@@ -429,6 +458,11 @@ func (s) TestDelegatingResolverForEndpointsWithProxy(t *testing.T) {
 	case <-time.After(defaultTestShortTimeout):
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
 	proxyResolver.UpdateState(resolver.State{
 		Endpoints: []resolver.Endpoint{
 			{Addresses: []resolver.Address{{Addr: resolvedProxyTestAddr1}}},
@@ -460,8 +494,8 @@ func (s) TestDelegatingResolverForEndpointsWithProxy(t *testing.T) {
 	var gotState resolver.State
 	select {
 	case gotState = <-stateCh:
-	case <-time.After(defaultTestTimeout):
-		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	case <-ctx.Done():
+		t.Fatal("Contex timed out when waiting for a state update from the delegating resolver")
 	}
 
 	if diff := cmp.Diff(gotState, wantState); diff != "" {
@@ -503,8 +537,7 @@ func (s) TestDelegatingResolverForMultipleProxyAddress(t *testing.T) {
 	targetResolver := manual.NewBuilderWithScheme("test")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
-
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
@@ -524,6 +557,11 @@ func (s) TestDelegatingResolverForMultipleProxyAddress(t *testing.T) {
 	case <-time.After(defaultTestShortTimeout):
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
 	proxyResolver.UpdateState(resolver.State{
 		Addresses: []resolver.Address{
 			{Addr: resolvedProxyTestAddr1},
@@ -542,8 +580,8 @@ func (s) TestDelegatingResolverForMultipleProxyAddress(t *testing.T) {
 	var gotState resolver.State
 	select {
 	case gotState = <-stateCh:
-	case <-time.After(defaultTestTimeout):
-		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	case <-ctx.Done():
+		t.Fatal("Context timed out when waiting for a state update from the delegating resolver")
 	}
 
 	if diff := cmp.Diff(gotState, wantState); diff != "" {
@@ -586,12 +624,8 @@ func (s) TestDelegatingResolverUpdateStateDuringClose(t *testing.T) {
 
 	target := targetResolver.Scheme() + ":///" + "ignored"
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
-	proxyResolverBuilt := make(chan struct{})
-	proxyResolver.BuildCallback = func(resolver.Target, resolver.ClientConn, resolver.BuildOptions) {
-		close(proxyResolverBuilt)
-	}
 	unblockProxyResolverClose := make(chan struct{}, 1)
 	proxyResolver.CloseCallback = func() {
 		<-unblockProxyResolverClose
@@ -614,11 +648,7 @@ func (s) TestDelegatingResolverUpdateStateDuringClose(t *testing.T) {
 	// Wait for the proxy resolver to be built before calling Close.
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	select {
-	case <-proxyResolverBuilt:
-	case <-ctx.Done():
-		t.Fatalf("Context timed out waiting for proxy resolver to be built.")
-	}
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
 	// Closing the delegating resolver will block until the test writes to the
 	// unblockProxyResolverClose channel.
 	go dr.Close()
@@ -677,7 +707,7 @@ func (s) TestDelegatingResolverUpdateStateFromResolveNow(t *testing.T) {
 
 	target := targetResolver.Scheme() + ":///" + "ignored"
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
 	tcc, _, _ := createTestResolverClientConn(t)
 	tcc.UpdateStateF = func(resolver.State) error {
@@ -692,14 +722,18 @@ func (s) TestDelegatingResolverUpdateStateFromResolveNow(t *testing.T) {
 		Endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "1.1.1.1"}}}},
 	})
 
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
+
 	// Updating the channel will result in an error being returned. The
 	// delegating resolver should call call "ResolveNow" on the target resolver.
 	proxyResolver.UpdateState(resolver.State{
 		Endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "1.1.1.1"}}}},
 	})
 
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	select {
 	case <-targetResolverCalled:
 	case <-ctx.Done():
@@ -737,12 +771,9 @@ func (s) TestDelegatingResolverResolveNow(t *testing.T) {
 
 	target := targetResolver.Scheme() + ":///" + "ignored"
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
+
 	proxyResolverCalled := make(chan struct{})
-	proxyResolverBuilt := make(chan struct{})
-	proxyResolver.BuildCallback = func(resolver.Target, resolver.ClientConn, resolver.BuildOptions) {
-		close(proxyResolverBuilt)
-	}
 	proxyResolver.ResolveNowCallback = func(resolver.ResolveNowOptions) {
 		// Updating the resolver state should not deadlock.
 		proxyResolver.CC().UpdateState(resolver.State{
@@ -770,12 +801,7 @@ func (s) TestDelegatingResolverResolveNow(t *testing.T) {
 		t.Fatalf("context timed out waiting for targetResolver.ResolveNow() to be called.")
 	}
 
-	// Wait for proxy resolver to be built.
-	select {
-	case <-proxyResolverBuilt:
-	case <-ctx.Done():
-		t.Fatalf("Timeout when waiting for proxy resolver to be built")
-	}
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
 
 	dr.ResolveNow(resolver.ResolveNowOptions{})
 
@@ -821,11 +847,7 @@ func (s) TestDelegatingResolverForNonTCPTarget(t *testing.T) {
 	targetResolver := manual.NewBuilderWithScheme("test")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
-	proxyBuildCalled := make(chan struct{})
-	proxyResolver.BuildCallback = func(resolver.Target, resolver.ClientConn, resolver.BuildOptions) {
-		close(proxyBuildCalled)
-	}
+	_, proxyResolverBuilt := setupDNS(t)
 
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
@@ -850,7 +872,7 @@ func (s) TestDelegatingResolverForNonTCPTarget(t *testing.T) {
 	// Verify that the delegating resolver doesn't call proxy resolver's
 	// UpdateState since we have no tcp address
 	select {
-	case <-proxyBuildCalled:
+	case <-proxyResolverBuilt:
 		t.Fatal("Unexpected call to proxy resolver update state")
 	case <-time.After(defaultTestShortTimeout):
 	}
@@ -900,7 +922,7 @@ func (s) TestDelegatingResolverForMixNetworkType(t *testing.T) {
 	targetResolver := manual.NewBuilderWithScheme("test")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
@@ -920,6 +942,11 @@ func (s) TestDelegatingResolverForMixNetworkType(t *testing.T) {
 	case <-time.After(defaultTestShortTimeout):
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
 	proxyResolver.UpdateState(resolver.State{
 		Addresses:     []resolver.Address{{Addr: envProxyAddr}},
 		ServiceConfig: &serviceconfig.ParseResult{},
@@ -928,8 +955,8 @@ func (s) TestDelegatingResolverForMixNetworkType(t *testing.T) {
 	var gotState resolver.State
 	select {
 	case gotState = <-stateCh:
-	case <-time.After(defaultTestTimeout):
-		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	case <-ctx.Done():
+		t.Fatal("Context timed out when waiting for a state update from the delegating resolver")
 	}
 	wantState := resolver.State{
 		Addresses:     []resolver.Address{nonTCPAddr, proxyAddressWithTargetAttribute(envProxyAddr, resolvedTargetTestAddr2)},
diff --git a/internal/stats/metrics_recorder_list_test.go b/internal/stats/metrics_recorder_list_test.go
index fef3d1df..38f9472f 100644
--- a/internal/stats/metrics_recorder_list_test.go
+++ b/internal/stats/metrics_recorder_list_test.go
@@ -23,7 +23,6 @@ package stats_test
 import (
 	"context"
 	"fmt"
-	"log"
 	"strings"
 	"testing"
 	"time"
@@ -154,7 +153,7 @@ func (s) TestMetricsRecorderList(t *testing.T) {
 
 	cc, err := grpc.NewClient(mr.Scheme()+":///", grpc.WithResolvers(mr), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithStatsHandler(mr2))
 	if err != nil {
-		log.Fatalf("Failed to dial: %v", err)
+		t.Fatalf("grpc.NewClient() failed: %v", err)
 	}
 	defer cc.Close()
 
diff --git a/internal/transport/http_util_test.go b/internal/transport/http_util_test.go
index 5eb466f3..eaf17d0c 100644
--- a/internal/transport/http_util_test.go
+++ b/internal/transport/http_util_test.go
@@ -22,28 +22,49 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"reflect"
 	"testing"
 	"time"
 )
 
-func (s) TestTimeoutDecode(t *testing.T) {
+func (s) TestDecodeTimeout(t *testing.T) {
 	for _, test := range []struct {
 		// input
 		s string
 		// output
-		d   time.Duration
-		err error
+		d       time.Duration
+		wantErr bool
 	}{
-		{"1234S", time.Second * 1234, nil},
-		{"1234x", 0, fmt.Errorf("transport: timeout unit is not recognized: %q", "1234x")},
-		{"1", 0, fmt.Errorf("transport: timeout string is too short: %q", "1")},
-		{"", 0, fmt.Errorf("transport: timeout string is too short: %q", "")},
+
+		{"00000001n", time.Nanosecond, false},
+		{"10u", time.Microsecond * 10, false},
+		{"00000010m", time.Millisecond * 10, false},
+		{"1234S", time.Second * 1234, false},
+		{"00000001M", time.Minute, false},
+		{"09999999S", time.Second * 9999999, false},
+		{"99999999S", time.Second * 99999999, false},
+		{"99999999M", time.Minute * 99999999, false},
+		{"2562047H", time.Hour * 2562047, false},
+		{"2562048H", time.Duration(math.MaxInt64), false},
+		{"99999999H", time.Duration(math.MaxInt64), false},
+		{"-1S", 0, true},
+		{"1234x", 0, true},
+		{"1234s", 0, true},
+		{"1234", 0, true},
+		{"1", 0, true},
+		{"", 0, true},
+		{"9a1S", 0, true},
+		{"0S", 0, true}, // PROTOCOL-HTTP2.md requires positive integers
+		{"00000000S", 0, true},
+		{"000000000S", 0, true},
 	} {
 		d, err := decodeTimeout(test.s)
-		if d != test.d || fmt.Sprint(err) != fmt.Sprint(test.err) {
-			t.Fatalf("timeoutDecode(%q) = %d, %v, want %d, %v", test.s, int64(d), err, int64(test.d), test.err)
+		gotErr := err != nil
+		if d != test.d || gotErr != test.wantErr {
+			t.Errorf("timeoutDecode(%q) = %d, %v, want %d, wantErr=%v",
+				test.s, int64(d), err, int64(test.d), test.wantErr)
 		}
 	}
 }
diff --git a/internal/transport/transport_test.go b/internal/transport/transport_test.go
index 8b121959..d2c4fac3 100644
--- a/internal/transport/transport_test.go
+++ b/internal/transport/transport_test.go
@@ -919,8 +919,9 @@ func (s) TestLargeMessageSuspension(t *testing.T) {
 	}
 	// The server will send an RST stream frame on observing the deadline
 	// expiration making the client stream fail with a DeadlineExceeded status.
-	if _, err := s.readTo(make([]byte, 8)); err != io.EOF {
-		t.Fatalf("Read got unexpected error: %v, want %v", err, io.EOF)
+	_, err = s.readTo(make([]byte, 8))
+	if st, ok := status.FromError(err); !ok || st.Code() != codes.DeadlineExceeded {
+		t.Fatalf("Read got unexpected error: %v, want status with code %v", err, codes.DeadlineExceeded)
 	}
 	if got, want := s.Status().Code(), codes.DeadlineExceeded; got != want {
 		t.Fatalf("Read got status %v with code %v, want %v", s.Status(), got, want)
diff --git a/stats/stats_test.go b/stats/stats_test.go
index ec5ffa04..c83e12f8 100644
--- a/stats/stats_test.go
+++ b/stats/stats_test.go
@@ -30,6 +30,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/grpctest"
@@ -268,13 +269,12 @@ func (te *test) startServer(ts testgrpc.TestServiceServer) {
 	te.srvAddr = lis.Addr().String()
 }
 
-func (te *test) clientConn() *grpc.ClientConn {
+func (te *test) clientConn(ctx context.Context) *grpc.ClientConn {
 	if te.cc != nil {
 		return te.cc
 	}
 	opts := []grpc.DialOption{
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithBlock(),
 		grpc.WithUserAgent("test/0.0.1"),
 	}
 	if te.compress == "gzip" {
@@ -288,10 +288,12 @@ func (te *test) clientConn() *grpc.ClientConn {
 	}
 
 	var err error
-	te.cc, err = grpc.Dial(te.srvAddr, opts...)
+	te.cc, err = grpc.NewClient(te.srvAddr, opts...)
 	if err != nil {
-		te.t.Fatalf("Dial(%q) = %v", te.srvAddr, err)
+		te.t.Fatalf("grpc.NewClient(%q) failed: %v", te.srvAddr, err)
 	}
+	te.cc.Connect()
+	testutils.AwaitState(ctx, te.t, te.cc, connectivity.Ready)
 	return te.cc
 }
 
@@ -317,15 +319,15 @@ func (te *test) doUnaryCall(c *rpcConfig) (*testpb.SimpleRequest, *testpb.Simple
 		req  *testpb.SimpleRequest
 		err  error
 	)
-	tc := testgrpc.NewTestServiceClient(te.clientConn())
+	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	tc := testgrpc.NewTestServiceClient(te.clientConn(tCtx))
 	if c.success {
 		req = &testpb.SimpleRequest{Payload: idToPayload(errorID + 1)}
 	} else {
 		req = &testpb.SimpleRequest{Payload: idToPayload(errorID)}
 	}
 
-	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	resp, err = tc.UnaryCall(metadata.NewOutgoingContext(tCtx, testMetadata), req, grpc.WaitForReady(!c.failfast))
 	return req, resp, err
 }
@@ -336,9 +338,9 @@ func (te *test) doFullDuplexCallRoundtrip(c *rpcConfig) ([]proto.Message, []prot
 		resps []proto.Message
 		err   error
 	)
-	tc := testgrpc.NewTestServiceClient(te.clientConn())
 	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
+	tc := testgrpc.NewTestServiceClient(te.clientConn(tCtx))
 	stream, err := tc.FullDuplexCall(metadata.NewOutgoingContext(tCtx, testMetadata), grpc.WaitForReady(!c.failfast))
 	if err != nil {
 		return reqs, resps, err
@@ -377,9 +379,9 @@ func (te *test) doClientStreamCall(c *rpcConfig) ([]proto.Message, *testpb.Strea
 		resp *testpb.StreamingInputCallResponse
 		err  error
 	)
-	tc := testgrpc.NewTestServiceClient(te.clientConn())
 	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
+	tc := testgrpc.NewTestServiceClient(te.clientConn(tCtx))
 	stream, err := tc.StreamingInputCall(metadata.NewOutgoingContext(tCtx, testMetadata), grpc.WaitForReady(!c.failfast))
 	if err != nil {
 		return reqs, resp, err
@@ -407,16 +409,15 @@ func (te *test) doServerStreamCall(c *rpcConfig) (*testpb.StreamingOutputCallReq
 		resps []proto.Message
 		err   error
 	)
-
-	tc := testgrpc.NewTestServiceClient(te.clientConn())
+	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	tc := testgrpc.NewTestServiceClient(te.clientConn(tCtx))
 
 	var startID int32
 	if !c.success {
 		startID = errorID
 	}
 	req = &testpb.StreamingOutputCallRequest{Payload: idToPayload(startID)}
-	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	stream, err := tc.StreamingOutputCall(metadata.NewOutgoingContext(tCtx, testMetadata), req, grpc.WaitForReady(!c.failfast))
 	if err != nil {
 		return req, resps, err
diff --git a/test/channelz_test.go b/test/channelz_test.go
index 0a9b88bd..93847126 100644
--- a/test/channelz_test.go
+++ b/test/channelz_test.go
@@ -304,11 +304,11 @@ func (s) TestCZTopChannelRegistrationAndDeletion(t *testing.T) {
 	}
 }
 
-func (s) TestCZTopChannelRegistrationAndDeletionWhenDialFail(t *testing.T) {
-	// Make dial fails (due to no transport security specified)
+func (s) TestCZTopChannelRegistrationAndDeletionWhenNewClientFail(t *testing.T) {
+	// Make newclient fails (due to no transport security specified)
 	_, err := grpc.NewClient("fake.addr")
 	if err == nil {
-		t.Fatal("expecting dial to fail")
+		t.Fatal("expecting newclient to fail")
 	}
 	if tcs, end := channelz.GetTopChannels(0, 0); tcs != nil || !end {
 		t.Fatalf("GetTopChannels(0, 0) = %v, %v, want <nil>, true", tcs, end)
diff --git a/test/end2end_test.go b/test/end2end_test.go
index a4258771..fd5e29ea 100644
--- a/test/end2end_test.go
+++ b/test/end2end_test.go
@@ -854,10 +854,11 @@ func (te *test) clientConn(opts ...grpc.DialOption) *grpc.ClientConn {
 	var scheme string
 	opts, scheme = te.configDial(opts...)
 	var err error
-	te.cc, err = grpc.Dial(scheme+te.srvAddr, opts...)
+	te.cc, err = grpc.NewClient(scheme+te.srvAddr, opts...)
 	if err != nil {
-		te.t.Fatalf("Dial(%q) = %v", scheme+te.srvAddr, err)
+		te.t.Fatalf("grpc.NewClient(%q) failed: %v", scheme+te.srvAddr, err)
 	}
+	te.cc.Connect()
 	return te.cc
 }
 
diff --git a/test/transport_test.go b/test/transport_test.go
index 2f61a6c0..10e9ab57 100644
--- a/test/transport_test.go
+++ b/test/transport_test.go
@@ -19,16 +19,20 @@ package test
 
 import (
 	"context"
+	"encoding/binary"
 	"io"
 	"net"
 	"sync"
 	"testing"
 
+	"golang.org/x/net/http2"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal/grpcsync"
 	"google.golang.org/grpc/internal/stubserver"
+	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/internal/transport"
 	"google.golang.org/grpc/status"
 
@@ -153,3 +157,75 @@ func (s) TestClientTransportRestartsAfterStreamIDExhausted(t *testing.T) {
 		t.Fatal("Timeout expired when waiting for first client transport to close")
 	}
 }
+
+// Tests that an RST_STREAM frame that causes an io.ErrUnexpectedEOF while
+// reading a gRPC message is correctly converted to a gRPC status with code
+// CANCELLED. The test sends a data frame with a partial gRPC message, followed
+// by an RST_STREAM frame with HTTP/2 code CANCELLED. The test asserts the
+// client receives the correct status.
+func (s) TestRSTDuringMessageRead(t *testing.T) {
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer lis.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%s) = %v", lis.Addr().String(), err)
+	}
+	defer cc.Close()
+
+	go func() {
+		conn, err := lis.Accept()
+		if err != nil {
+			t.Errorf("lis.Accept() = %v", err)
+			return
+		}
+		defer conn.Close()
+		framer := http2.NewFramer(conn, conn)
+
+		if _, err := io.ReadFull(conn, make([]byte, len(clientPreface))); err != nil {
+			t.Errorf("Error while reading client preface: %v", err)
+			return
+		}
+		if err := framer.WriteSettings(); err != nil {
+			t.Errorf("Error while writing settings: %v", err)
+			return
+		}
+		if err := framer.WriteSettingsAck(); err != nil {
+			t.Errorf("Error while writing settings: %v", err)
+			return
+		}
+		for ctx.Err() == nil {
+			frame, err := framer.ReadFrame()
+			if err != nil {
+				return
+			}
+			switch frame := frame.(type) {
+			case *http2.HeadersFrame:
+				// When the client creates a stream, write a partial gRPC
+				// message followed by an RST_STREAM.
+				const messageLen = 2048
+				buf := make([]byte, messageLen/2)
+				// Write the gRPC message length header.
+				binary.BigEndian.PutUint32(buf[1:5], uint32(messageLen))
+				if err := framer.WriteData(1, false, buf); err != nil {
+					return
+				}
+				framer.WriteRSTStream(1, http2.ErrCodeCancel)
+			default:
+				t.Logf("Server received frame: %v", frame)
+			}
+		}
+	}()
+
+	// The server will send a partial gRPC message before cancelling the stream.
+	// The client should get a gRPC status with code CANCELLED.
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.Canceled {
+		t.Fatalf("client.EmptyCall() returned %v; want status with code %v", err, codes.Canceled)
+	}
+}