retry: prevent per-RPC creds error from being transparently retried (#…

…3677)

balancer/grpclb/grpclb_test.go

@@ -1351,9 +1351,9 @@ func (s) TestGRPCLBStatsUnaryFailedToSend(t *testing.T) {
 			cc.Invoke(context.Background(), failtosendURI, &testpb.Empty{}, nil)
 		}
 	}, &rpcStats{
-		numCallsStarted:                        int64(countRPC)*2 - 1,
-		numCallsFinished:                       int64(countRPC)*2 - 1,
-		numCallsFinishedWithClientFailedToSend: int64(countRPC-1) * 2,
+		numCallsStarted:                        int64(countRPC),
+		numCallsFinished:                       int64(countRPC),
+		numCallsFinishedWithClientFailedToSend: int64(countRPC) - 1,
 		numCallsFinishedKnownReceived:          1,
 	}); err != nil {
 		t.Fatal(err)
@@ -1444,9 +1444,9 @@ func (s) TestGRPCLBStatsStreamingFailedToSend(t *testing.T) {
 			cc.NewStream(context.Background(), &grpc.StreamDesc{}, failtosendURI)
 		}
 	}, &rpcStats{
-		numCallsStarted:                        int64(countRPC)*2 - 1,
-		numCallsFinished:                       int64(countRPC)*2 - 1,
-		numCallsFinishedWithClientFailedToSend: int64(countRPC-1) * 2,
+		numCallsStarted:                        int64(countRPC),
+		numCallsFinished:                       int64(countRPC),
+		numCallsFinishedWithClientFailedToSend: int64(countRPC) - 1,
 		numCallsFinishedKnownReceived:          1,
 	}); err != nil {
 		t.Fatal(err)

stream.go

@@ -459,26 +459,18 @@ func (cs *clientStream) commitAttempt() {
 // shouldRetry returns nil if the RPC should be retried; otherwise it returns
 // the error that should be returned by the operation.
 func (cs *clientStream) shouldRetry(err error) error {
-	if cs.attempt.s == nil && !cs.callInfo.failFast {
-		// In the event of any error from NewStream (attempt.s == nil), we
-		// never attempted to write anything to the wire, so we can retry
-		// indefinitely for non-fail-fast RPCs.
-		return nil
-	}
 	if cs.finished || cs.committed {
 		// RPC is finished or committed; cannot retry.
 		return err
 	}
 	// Wait for the trailers.
 	if cs.attempt.s != nil {
 		<-cs.attempt.s.Done()
+		if cs.firstAttempt && cs.attempt.s.Unprocessed() {
+			// First attempt, stream unprocessed: transparently retry.
+			return nil
+		}
 	}
-	if cs.firstAttempt && (cs.attempt.s == nil || cs.attempt.s.Unprocessed()) {
-		// First attempt, stream unprocessed: transparently retry.
-		cs.firstAttempt = false
-		return nil
-	}
-	cs.firstAttempt = false
 	if cs.cc.dopts.disableRetry {
 		return err
 	}
@@ -564,6 +556,7 @@ func (cs *clientStream) retryLocked(lastErr error) error {
 			cs.commitAttemptLocked()
 			return err
 		}
+		cs.firstAttempt = false
 		if err := cs.newAttemptLocked(nil, nil); err != nil {
 			return err
 		}

test/creds_test.go

@@ -189,11 +189,15 @@ func (s) TestGRPCMethodAccessibleToCredsViaContextRequestInfo(t *testing.T) {
 	cc := te.clientConn(grpc.WithPerRPCCredentials(&methodTestCreds{}))
 	tc := testpb.NewTestServiceClient(cc)
 
-	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()
 	if _, err := tc.EmptyCall(ctx, &testpb.Empty{}); status.Convert(err).Message() != wantMethod {
 		t.Fatalf("ss.client.EmptyCall(_, _) = _, %v; want _, _.Message()=%q", err, wantMethod)
 	}
+
+	if _, err := tc.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); status.Convert(err).Message() != wantMethod {
+		t.Fatalf("ss.client.EmptyCall(_, _) = _, %v; want _, _.Message()=%q", err, wantMethod)
+	}
 }
 
 const clientAlwaysFailCredErrorMsg = "clientAlwaysFailCred always fails"