Repository: grpc/grpc-go
File: stream.go (Go)

Code before:
		}
		if err != nil && err != io.EOF {
			st, _ := status.FromError(toRPCErr(err))
			ss.s.WriteStatus(st)
			// Non-user specified status was sent out. This should be an error
			// case (as a server side Cancel maybe).
			//
			// This is not handled specifically now. User will return a final
			// status from the service handler, we will log that error instead.
			// This behavior is similar to an interceptor.
		}
	}()
	var payInfo *payloadInfo
	if len(ss.statsHandler) != 0 || len(ss.binlogs) != 0 {
		payInfo = &payloadInfo{}
		defer payInfo.free()
	}
	if err := recv(ss.p, ss.codec, ss.s, ss.decompressorV0, m, ss.maxReceiveMessageSize, payInfo, ss.decompressorV1, true); err != nil {
		if err == io.EOF {
			if len(ss.binlogs) != 0 {
				chc := &binarylog.ClientHalfClose{}
				for _, binlog := range ss.binlogs {
					binlog.Log(ss.ctx, chc)
				}
			}
			if !ss.desc.ClientStreams {
				return status.Errorf(codes.Internal, "RecvMsg is called twice")
			}
			return err
		}
		if err == io.ErrUnexpectedEOF {
			err = status.Error(codes.Internal, io.ErrUnexpectedEOF.Error())
		}
		return toRPCErr(err)
	}
	if len(ss.statsHandler) != 0 {
		for _, sh := range ss.statsHandler {
			sh.HandleRPC(ss.s.Context(), &stats.InPayload{
				RecvTime:         time.Now(),
				Payload:          m,
				Length:           payInfo.uncompressedBytes.Len(),
				WireLength:       payInfo.compressedLength + headerLen,
				CompressedLength: payInfo.compressedLength,
			})
		}
	}
	if len(ss.binlogs) != 0 {
		cm := &binarylog.ClientMessage{
			Message: payInfo.uncompressedBytes.Materialize(),
		}
		for _, binlog := range ss.binlogs {

Review comment (suggestion):
It's not immediately clear to me how this conditions indicates RecvMsg is called twice. From my understanding, the handling of cardinality violations in the server stream should be similar to the one in client stream. The check here should be for 0 requests being received for a non-client-streaming request.
https://github.com/grpc/grpc-go/blob/e5de1e2cac527d419f53fbe4ae07628c688ff0dd/stream.go#L1142-L1144

For > 1 requests, there should be a block near the end of this method similar to the following:
https://github.com/grpc/grpc-go/blob/e5de1e2cac527d419f53fbe4ae07628c688ff0dd/stream.go#L1167-L1178

Code after:
			}
			ss.mu.Unlock()
		}
		if err != nil && err != io.EOF {
			st, _ := status.FromError(toRPCErr(err))
			ss.s.WriteStatus(st)
			// Non-user specified status was sent out. This should be an error
			// case (as a server side Cancel maybe).
			//
			// This is not handled specifically now. User will return a final
			// status from the service handler, we will log that error instead.
			// This behavior is similar to an interceptor.
		}
	}()
	var payInfo *payloadInfo
	if len(ss.statsHandler) != 0 || len(ss.binlogs) != 0 {
		payInfo = &payloadInfo{}
		defer payInfo.free()
	}
	if err := recv(ss.p, ss.codec, ss.s, ss.decompressorV0, m, ss.maxReceiveMessageSize, payInfo, ss.decompressorV1, true); err != nil {
		if err == io.EOF {
			if len(ss.binlogs) != 0 {
				chc := &binarylog.ClientHalfClose{}
				for _, binlog := range ss.binlogs {
					binlog.Log(ss.ctx, chc)
				}
			}
			// Received no request msg for non-client streaming rpcs.
			if !ss.desc.ClientStreams && !ss.recvFirstMsg {
				return status.Error(codes.Internal, "cardinality violation: received no request message from non-client-streaming RPC")
			}
			return err
		}
		if err == io.ErrUnexpectedEOF {
			err = status.Error(codes.Internal, io.ErrUnexpectedEOF.Error())
		}
		return toRPCErr(err)
	}
	ss.recvFirstMsg = true
	if len(ss.statsHandler) != 0 {
		for _, sh := range ss.statsHandler {
			sh.HandleRPC(ss.s.Context(), &stats.InPayload{
				RecvTime:         time.Now(),
				Payload:          m,
				Length:           payInfo.uncompressedBytes.Len(),
				WireLength:       payInfo.compressedLength + headerLen,
				CompressedLength: payInfo.compressedLength,
			})
		}
	}
	if len(ss.binlogs) != 0 {