grpc: Fix cardinality violations in non-server streaming RPCs

stream.go

@@ -542,8 +542,6 @@ type clientStream struct {
 
 	sentLast bool // sent an end stream
 
-	recvFirstMsg bool // received first msg from server
-
 	methodConfig *MethodConfig
 
 	ctx context.Context // the application's context, wrapped by stats/tracing
@@ -1137,17 +1135,14 @@ func (a *csAttempt) recvMsg(m any, payInfo *payloadInfo) (err error) {
 				return statusErr
 			}
 			// received no msg and status ok for non-server streaming rpcs.
-			if !cs.desc.ServerStreams && !cs.recvFirstMsg {
+			if !cs.desc.ServerStreams {
 				return status.Errorf(codes.Internal, "client streaming cardinality violation")
 			}
 			return io.EOF // indicates successful end of stream.
 		}
 
 		return toRPCErr(err)
 	}
-	if !cs.desc.ServerStreams {
-		cs.recvFirstMsg = true
-	}
 	if a.trInfo != nil {
 		a.mu.Lock()
 		if a.trInfo.tr != nil {

test/end2end_test.go

@@ -3736,6 +3736,87 @@ func (s) TestClientStreaming_ReturnErrorAfterSendAndClose(t *testing.T) {
 	}
 }
 
+// Tests that a client receives a cardinality violation error for unary
+// RPCs if the server doesn't send a message before returning status OK.
+func (s) TestUnaryRPC_ServerSendsOnlyTrailersWithOK(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 reading client preface: %v", err)
+			return
+		}
+		if err := framer.WriteSettings(); err != nil {
+			t.Errorf("Error writing server settings: %v", err)
+			return
+		}
+		if err := framer.WriteSettingsAck(); err != nil {
+			t.Errorf("Error writing settings ack: %v", err)
+			return
+		}
+
+		for ctx.Err() == nil {
+			frame, err := framer.ReadFrame()
+			if err != nil {
+				t.Errorf("Error reading frame: %v", err)
+				return
+			}
+
+			switch frame := frame.(type) {
+			case *http2.HeadersFrame:
+				if frame.Header().StreamID != 1 {
+					t.Errorf("Expected stream ID 1, got %d", frame.Header().StreamID)
+					return
+				}
+
+				var buf bytes.Buffer
+				enc := hpack.NewEncoder(&buf)
+				_ = enc.WriteField(hpack.HeaderField{Name: ":status", Value: "200"})
+				_ = enc.WriteField(hpack.HeaderField{Name: "content-type", Value: "application/grpc"})
+				_ = enc.WriteField(hpack.HeaderField{Name: "grpc-status", Value: "0"})
+
+				if err := framer.WriteHeaders(http2.HeadersFrameParam{
+					StreamID:      1,
+					BlockFragment: buf.Bytes(),
+					EndHeaders:    true,
+					EndStream:     true,
+				}); err != nil {
+					t.Errorf("Error while writing headers: %v", err)
+					return
+				}
+				time.Sleep(50 * time.Millisecond)
+			default:
+				t.Logf("Server received frame: %v", frame)
+			}
+		}
+	}()
+
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err = client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.Internal {
+		t.Errorf("stream.RecvMsg() = %v, want error %v", status.Code(err), codes.Internal)
+	}
+}
+
 func (s) TestExceedMaxStreamsLimit(t *testing.T) {
 	for _, e := range listTestEnv() {
 		testExceedMaxStreamsLimit(t, e)