Skip to content

Commit 42a184d

Browse files
committed
revert commit 20bd1e7
1 parent 206ddf8 commit 42a184d

2 files changed

Lines changed: 116 additions & 6 deletions

File tree

stream.go

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1144,6 +1144,10 @@ func (a *csAttempt) recvMsg(m any, payInfo *payloadInfo) (err error) {
11441144
if statusErr := a.transportStream.Status().Err(); statusErr != nil {
11451145
return statusErr
11461146
}
1147+
// Received no msg and status OK for non-server streaming rpcs.
1148+
if !cs.desc.ServerStreams {
1149+
return status.Error(codes.Internal, "cardinality violation: received no response message from non-streaming RPC")
1150+
}
11471151
return io.EOF // indicates successful end of stream.
11481152
}
11491153

@@ -1177,7 +1181,7 @@ func (a *csAttempt) recvMsg(m any, payInfo *payloadInfo) (err error) {
11771181
} else if err != nil {
11781182
return toRPCErr(err)
11791183
}
1180-
return status.Errorf(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
1184+
return status.Error(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
11811185
}
11821186

11831187
func (a *csAttempt) finish(err error) {
@@ -1484,6 +1488,10 @@ func (as *addrConnStream) RecvMsg(m any) (err error) {
14841488
if statusErr := as.transportStream.Status().Err(); statusErr != nil {
14851489
return statusErr
14861490
}
1491+
// Received no msg and status OK for non-server streaming rpcs.
1492+
if !as.desc.ServerStreams {
1493+
return status.Error(codes.Internal, "cardinality violation: received no response message from non-streaming RPC")
1494+
}
14871495
return io.EOF // indicates successful end of stream.
14881496
}
14891497
return toRPCErr(err)
@@ -1501,7 +1509,7 @@ func (as *addrConnStream) RecvMsg(m any) (err error) {
15011509
} else if err != nil {
15021510
return toRPCErr(err)
15031511
}
1504-
return status.Errorf(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
1512+
return status.Error(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
15051513
}
15061514

15071515
func (as *addrConnStream) finish(err error) {

test/end2end_test.go

Lines changed: 106 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3590,9 +3590,6 @@ func testClientStreamingError(t *testing.T, e env) {
35903590
// Tests that a client receives a cardinality violation error for client-streaming
35913591
// RPCs if the server doesn't send a message before returning status OK.
35923592
func (s) TestClientStreamingCardinalityViolation_ServerHandlerMissingSendAndClose(t *testing.T) {
3593-
// TODO : https://github.com/grpc/grpc-go/issues/8119 - remove `t.Skip()`
3594-
// after this is fixed.
3595-
t.Skip()
35963593
ss := &stubserver.StubServer{
35973594
StreamingInputCallF: func(_ testgrpc.TestService_StreamingInputCallServer) error {
35983595
// Returning status OK without sending a response message.This is a
@@ -3741,6 +3738,111 @@ func (s) TestClientStreaming_ReturnErrorAfterSendAndClose(t *testing.T) {
37413738
}
37423739
}
37433740

3741+
// Tests that a client receives a cardinality violation error for unary
3742+
// RPCs if the server doesn't send a message before returning status OK.
3743+
func (s) TestUnaryRPC_ServerSendsOnlyTrailersWithOK(t *testing.T) {
3744+
lis, err := testutils.LocalTCPListener()
3745+
if err != nil {
3746+
t.Fatal(err)
3747+
}
3748+
defer lis.Close()
3749+
3750+
ss := grpc.UnknownServiceHandler(func(any, grpc.ServerStream) error {
3751+
return nil
3752+
})
3753+
3754+
s := grpc.NewServer(ss)
3755+
go s.Serve(lis)
3756+
defer s.Stop()
3757+
3758+
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
3759+
defer cancel()
3760+
cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
3761+
if err != nil {
3762+
t.Fatalf("grpc.NewClient(%q) failed unexpectedly: %v", lis.Addr(), err)
3763+
}
3764+
defer cc.Close()
3765+
3766+
client := testgrpc.NewTestServiceClient(cc)
3767+
if _, err = client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.Internal {
3768+
t.Errorf("stream.RecvMsg() = %v, want error %v", status.Code(err), codes.Internal)
3769+
}
3770+
}
3771+
3772+
// Tests that client will receive cardinality violations when calling
3773+
// RecvMsg() multiple times for non-streaming response streams.
3774+
func (s) TestUnaryRPC_ClientCallRecvMsgTwice(t *testing.T) {
3775+
e := tcpTLSEnv
3776+
te := newTest(t, e)
3777+
defer te.tearDown()
3778+
3779+
te.startServer(&testServer{security: e.security})
3780+
3781+
cc := te.clientConn()
3782+
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
3783+
defer cancel()
3784+
3785+
desc := &grpc.StreamDesc{
3786+
StreamName: "UnaryCall",
3787+
ServerStreams: false,
3788+
ClientStreams: false,
3789+
}
3790+
stream, err := cc.NewStream(ctx, desc, "/grpc.testing.TestService/UnaryCall")
3791+
if err != nil {
3792+
t.Fatalf("cc.NewStream() failed unexpectedly: %v", err)
3793+
}
3794+
3795+
if err := stream.SendMsg(&testpb.SimpleRequest{}); err != nil {
3796+
t.Fatalf("stream.SendMsg(_) = %v, want <nil>", err)
3797+
}
3798+
3799+
resp := &testpb.SimpleResponse{}
3800+
if err := stream.RecvMsg(resp); err != nil {
3801+
t.Fatalf("stream.RecvMsg() = %v , want <nil>", err)
3802+
}
3803+
3804+
if err = stream.RecvMsg(resp); status.Code(err) != codes.Internal {
3805+
t.Errorf("stream.RecvMsg() = %v, want error %v", status.Code(err), codes.Internal)
3806+
}
3807+
}
3808+
3809+
// Tests that client will receive cardinality violations when calling
3810+
// RecvMsg() multiple times for non-streaming response streams.
3811+
func (s) TestClientStreaming_ClientCallRecvMsgTwice(t *testing.T) {
3812+
ss := stubserver.StubServer{
3813+
StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {
3814+
if err := stream.SendAndClose(&testpb.StreamingInputCallResponse{}); err != nil {
3815+
t.Errorf("stream.SendAndClose(_) = %v, want <nil>", err)
3816+
}
3817+
return nil
3818+
},
3819+
}
3820+
if err := ss.Start(nil); err != nil {
3821+
t.Fatal("Error starting server:", err)
3822+
}
3823+
defer ss.Stop()
3824+
3825+
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
3826+
defer cancel()
3827+
stream, err := ss.Client.StreamingInputCall(ctx)
3828+
if err != nil {
3829+
t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
3830+
}
3831+
if err := stream.Send(&testpb.StreamingInputCallRequest{}); err != nil {
3832+
t.Fatalf("stream.Send(_) = %v, want <nil>", err)
3833+
}
3834+
if err := stream.CloseSend(); err != nil {
3835+
t.Fatalf("stream.CloseSend() = %v, want <nil>", err)
3836+
}
3837+
resp := new(testpb.StreamingInputCallResponse)
3838+
if err := stream.RecvMsg(resp); err != nil {
3839+
t.Fatalf("stream.RecvMsg() = %v , want <nil>", err)
3840+
}
3841+
if err = stream.RecvMsg(resp); status.Code(err) != codes.Internal {
3842+
t.Errorf("stream.RecvMsg() = %v, want error %v", status.Code(err), codes.Internal)
3843+
}
3844+
}
3845+
37443846
// Tests the behavior for server-side streaming when client calls SendMsg twice.
37453847
// Second call to SendMsg should fail with Internal error and result in closing
37463848
// the connection with a RST_STREAM.
@@ -3981,7 +4083,7 @@ func (s) TestServerStreaming_ClientSendsZeroRequests(t *testing.T) {
39814083
}
39824084

39834085
// Tests that a client receives a cardinality violation error for client-streaming
3984-
// RPCs if the server call SendMsg multiple times.
4086+
// RPCs if the server call SendMsg() multiple times.
39854087
func (s) TestClientStreaming_ServerHandlerSendMsgAfterSendMsg(t *testing.T) {
39864088
ss := stubserver.StubServer{
39874089
StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {

0 commit comments

Comments
 (0)