grpc: Fix cardinality violations in non-server streaming RPCs

stream.go

@@ -1134,6 +1134,10 @@ func (a *csAttempt) recvMsg(m any, payInfo *payloadInfo) (err error) {
 			if statusErr := a.transportStream.Status().Err(); statusErr != nil {
 				return statusErr
 			}
+			// received no msg and status ok for non-server streaming rpcs.
+			if !cs.desc.ServerStreams {
+				return status.Errorf(codes.Internal, "client streaming cardinality violation")
+			}
 			return io.EOF // indicates successful end of stream.
 		}
 

test/end2end_test.go

@@ -3588,9 +3588,6 @@ func testClientStreamingError(t *testing.T, e env) {
 // Tests that a client receives a cardinality violation error for client-streaming
 // RPCs if the server doesn't send a message before returning status OK.
 func (s) TestClientStreamingCardinalityViolation_ServerHandlerMissingSendAndClose(t *testing.T) {
-	// TODO : https://github.com/grpc/grpc-go/issues/8119 - remove `t.Skip()`
-	// after this is fixed.
-	t.Skip()
 	ss := &stubserver.StubServer{
 		StreamingInputCallF: func(_ testgrpc.TestService_StreamingInputCallServer) error {
 			// Returning status OK without sending a response message.This is a
@@ -3739,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)