Skip to content

Commit 206ddf8

Browse files
Pranjali-2501Pranjali-2501
committed
Revert "add extra state to track calls to client.recvmsg"
This reverts commit 6792a42.
1 parent 6792a42 commit 206ddf8

2 files changed

Lines changed: 7 additions & 175 deletions

File tree

‎stream.go‎

Lines changed: 2 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -549,8 +549,6 @@ type clientStream struct {
549549

550550
sentLast bool // sent an end stream
551551

552-
recvFirstMsg bool // set after the first message is received
553-
554552
methodConfig *MethodConfig
555553

556554
ctx context.Context // the application's context, wrapped by stats/tracing
@@ -1146,16 +1144,11 @@ func (a *csAttempt) recvMsg(m any, payInfo *payloadInfo) (err error) {
11461144
if statusErr := a.transportStream.Status().Err(); statusErr != nil {
11471145
return statusErr
11481146
}
1149-
// Received no msg and status OK for non-server streaming rpcs.
1150-
if !cs.desc.ServerStreams && !cs.recvFirstMsg {
1151-
return status.Error(codes.Internal, "cardinality violation: received no response message from non-server-streaming RPC")
1152-
}
11531147
return io.EOF // indicates successful end of stream.
11541148
}
11551149

11561150
return toRPCErr(err)
11571151
}
1158-
cs.recvFirstMsg = true
11591152
if a.trInfo != nil {
11601153
a.mu.Lock()
11611154
if a.trInfo.tr != nil {
@@ -1184,7 +1177,7 @@ func (a *csAttempt) recvMsg(m any, payInfo *payloadInfo) (err error) {
11841177
} else if err != nil {
11851178
return toRPCErr(err)
11861179
}
1187-
return status.Error(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
1180+
return status.Errorf(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
11881181
}
11891182

11901183
func (a *csAttempt) finish(err error) {
@@ -1366,7 +1359,6 @@ type addrConnStream struct {
13661359
transport transport.ClientTransport
13671360
ctx context.Context
13681361
sentLast bool
1369-
recvFirstMsg bool
13701362
desc *StreamDesc
13711363
codec baseCodec
13721364
sendCompressorV0 Compressor
@@ -1492,15 +1484,10 @@ func (as *addrConnStream) RecvMsg(m any) (err error) {
14921484
if statusErr := as.transportStream.Status().Err(); statusErr != nil {
14931485
return statusErr
14941486
}
1495-
// Received no msg and status OK for non-server streaming rpcs.
1496-
if !as.desc.ServerStreams && !as.recvFirstMsg {
1497-
return status.Error(codes.Internal, "cardinality violation: received no response message from non-server-streaming RPC")
1498-
}
14991487
return io.EOF // indicates successful end of stream.
15001488
}
15011489
return toRPCErr(err)
15021490
}
1503-
as.recvFirstMsg = true
15041491

15051492
if as.desc.ServerStreams {
15061493
// Subsequent messages should be received by subsequent RecvMsg calls.
@@ -1514,7 +1501,7 @@ func (as *addrConnStream) RecvMsg(m any) (err error) {
15141501
} else if err != nil {
15151502
return toRPCErr(err)
15161503
}
1517-
return status.Error(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
1504+
return status.Errorf(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
15181505
}
15191506

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

‎test/end2end_test.go‎

Lines changed: 5 additions & 160 deletions
Original file line numberDiff line numberDiff line change
@@ -3590,6 +3590,9 @@ 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()
35933596
ss := &stubserver.StubServer{
35943597
StreamingInputCallF: func(_ testgrpc.TestService_StreamingInputCallServer) error {
35953598
// Returning status OK without sending a response message.This is a
@@ -3738,165 +3741,6 @@ func (s) TestClientStreaming_ReturnErrorAfterSendAndClose(t *testing.T) {
37383741
}
37393742
}
37403743

3741-
// Tests that 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 the behavior for unary RPC when client calls RecvMsg() twice.
3773-
// Second call to RecvMsg should fail with io.EOF.
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); err != io.EOF {
3805-
t.Errorf("stream.RecvMsg() = %v, want error %v", err, io.EOF)
3806-
}
3807-
}
3808-
3809-
// Tests the behavior for unary RPC when server calls SendMsg() twice.
3810-
// Client should fail with cardinality violation error.
3811-
func (s) TestUnaryRPC_ServerCallSendMsgTwice(t *testing.T) {
3812-
lis, err := testutils.LocalTCPListener()
3813-
if err != nil {
3814-
t.Fatal(err)
3815-
}
3816-
defer lis.Close()
3817-
3818-
s := grpc.NewServer()
3819-
serviceDesc := grpc.ServiceDesc{
3820-
ServiceName: "grpc.testing.TestService",
3821-
HandlerType: (*any)(nil),
3822-
Methods: []grpc.MethodDesc{},
3823-
Streams: []grpc.StreamDesc{
3824-
{
3825-
StreamName: "UnaryCall",
3826-
Handler: func(_ any, stream grpc.ServerStream) error {
3827-
if err := stream.RecvMsg(&testpb.Empty{}); err != nil {
3828-
t.Errorf("stream.RecvMsg() = %v, want <nil>", err)
3829-
}
3830-
3831-
if err = stream.SendMsg(&testpb.Empty{}); err != nil {
3832-
t.Errorf("stream.SendMsg() = %v, want <nil>", err)
3833-
}
3834-
3835-
if err = stream.SendMsg(&testpb.Empty{}); err != nil {
3836-
t.Errorf("stream.SendMsg() = %v, want <nil>", err)
3837-
}
3838-
return nil
3839-
},
3840-
ClientStreams: false,
3841-
ServerStreams: false,
3842-
},
3843-
},
3844-
}
3845-
s.RegisterService(&serviceDesc, &testServer{})
3846-
go s.Serve(lis)
3847-
defer s.Stop()
3848-
3849-
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
3850-
defer cancel()
3851-
cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
3852-
if err != nil {
3853-
t.Fatalf("grpc.NewClient(%q) failed unexpectedly: %v", lis.Addr(), err)
3854-
}
3855-
defer cc.Close()
3856-
3857-
client := testgrpc.NewTestServiceClient(cc)
3858-
if _, err = client.UnaryCall(ctx, &testpb.SimpleRequest{}); status.Code(err) != codes.Internal {
3859-
t.Errorf("stream.RecvMsg() = %v, want error %v", status.Code(err), codes.Internal)
3860-
}
3861-
}
3862-
3863-
// Tests the behavior for client-streaming RPC when client calls RecvMsg() twice.
3864-
// Second call to RecvMsg should fail with io.EOF.
3865-
func (s) TestClientStreaming_ClientCallRecvMsgTwice(t *testing.T) {
3866-
ss := stubserver.StubServer{
3867-
StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {
3868-
if err := stream.SendAndClose(&testpb.StreamingInputCallResponse{}); err != nil {
3869-
t.Errorf("stream.SendAndClose(_) = %v, want <nil>", err)
3870-
}
3871-
return nil
3872-
},
3873-
}
3874-
if err := ss.Start(nil); err != nil {
3875-
t.Fatal("Error starting server:", err)
3876-
}
3877-
defer ss.Stop()
3878-
3879-
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
3880-
defer cancel()
3881-
stream, err := ss.Client.StreamingInputCall(ctx)
3882-
if err != nil {
3883-
t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
3884-
}
3885-
if err := stream.Send(&testpb.StreamingInputCallRequest{}); err != nil {
3886-
t.Fatalf("stream.Send(_) = %v, want <nil>", err)
3887-
}
3888-
if err := stream.CloseSend(); err != nil {
3889-
t.Fatalf("stream.CloseSend() = %v, want <nil>", err)
3890-
}
3891-
resp := new(testpb.StreamingInputCallResponse)
3892-
if err := stream.RecvMsg(resp); err != nil {
3893-
t.Fatalf("stream.RecvMsg() = %v , want <nil>", err)
3894-
}
3895-
if err = stream.RecvMsg(resp); err != io.EOF {
3896-
t.Errorf("stream.RecvMsg() = %v, want error %v", err, io.EOF)
3897-
}
3898-
}
3899-
39003744
// Tests the behavior for server-side streaming when client calls SendMsg twice.
39013745
// Second call to SendMsg should fail with Internal error and result in closing
39023746
// the connection with a RST_STREAM.
@@ -3958,6 +3802,7 @@ func (s) TestServerStreaming_ClientCallSendMsgTwice(t *testing.T) {
39583802
<-handlerDone
39593803
}
39603804

3805+
// TODO(i/7286) : Add tests to check server-side behavior for Unary RPC.
39613806
// Tests the behavior for unary RPC when client calls SendMsg twice. Second call
39623807
// to SendMsg should fail with Internal error.
39633808
func (s) TestUnaryRPC_ClientCallSendMsgTwice(t *testing.T) {
@@ -4136,7 +3981,7 @@ func (s) TestServerStreaming_ClientSendsZeroRequests(t *testing.T) {
41363981
}
41373982

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

0 commit comments

Comments
 (0)