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

Code before:
			a.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
		}
		a.mu.Unlock()
	}
	for _, sh := range a.statsHandlers {
		sh.HandleRPC(a.ctx, &stats.InPayload{
			Client:           true,
			RecvTime:         time.Now(),
			Payload:          m,
			WireLength:       payInfo.compressedLength + headerLen,
			CompressedLength: payInfo.compressedLength,
			Length:           payInfo.uncompressedBytes.Len(),
		})
	}
	if cs.desc.ServerStreams {
		// Subsequent messages should be received by subsequent RecvMsg calls.
		return nil
	}
	// Special handling for non-server-stream rpcs.
	// This recv expects EOF or errors, so we don't collect inPayload.
	if err := recv(a.parser, cs.codec, a.transportStream, a.decompressorV0, m, *cs.callInfo.maxReceiveMessageSize, nil, a.decompressorV1, false); err == io.EOF {
		return a.transportStream.Status().Err() // non-server streaming Recv returns nil on success
	} else if err != nil {
		return toRPCErr(err)
	}
	return status.Errorf(codes.Internal, "client streaming cardinality violation: get <nil>, want <EOF>")
}

func (a *csAttempt) finish(err error) {
	a.mu.Lock()
	if a.finished {
		a.mu.Unlock()
		return
	}
	a.finished = true
	if err == io.EOF {
		// Ending a stream with EOF indicates a success.
		err = nil
	}
	var tr metadata.MD
	if a.transportStream != nil {
		a.transportStream.Close(err)
		tr = a.transportStream.Trailer()
	}

	if a.pickResult.Done != nil {
		br := false
		if a.transportStream != nil {
			br = a.transportStream.BytesReceived()
		}
		a.pickResult.Done(balancer.DoneInfo{

Review comment (suggestion):
I think we can improve this error message to say something like:
`cardinality violation: expected EOF from non-streaming server, but received another message`. I believe this can also happen in unary RPCs.

Code after:
			a.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
		}
		a.mu.Unlock()
	}
	for _, sh := range a.statsHandlers {
		sh.HandleRPC(a.ctx, &stats.InPayload{
			Client:           true,
			RecvTime:         time.Now(),
			Payload:          m,
			WireLength:       payInfo.compressedLength + headerLen,
			CompressedLength: payInfo.compressedLength,
			Length:           payInfo.uncompressedBytes.Len(),
		})
	}
	if cs.desc.ServerStreams {
		// Subsequent messages should be received by subsequent RecvMsg calls.
		return nil
	}
	// Special handling for non-server-stream rpcs.
	// This recv expects EOF or errors, so we don't collect inPayload.
	if err := recv(a.parser, cs.codec, a.transportStream, a.decompressorV0, m, *cs.callInfo.maxReceiveMessageSize, nil, a.decompressorV1, false); err == io.EOF {
		return a.transportStream.Status().Err() // non-server streaming Recv returns nil on success
	} else if err != nil {
		return toRPCErr(err)
	}
	return status.Errorf(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
}

func (a *csAttempt) finish(err error) {
	a.mu.Lock()
	if a.finished {
		a.mu.Unlock()
		return
	}
	a.finished = true
	if err == io.EOF {
		// Ending a stream with EOF indicates a success.
		err = nil
	}
	var tr metadata.MD
	if a.transportStream != nil {
		a.transportStream.Close(err)
		tr = a.transportStream.Trailer()
	}

	if a.pickResult.Done != nil {
		br := false
		if a.transportStream != nil {
			br = a.transportStream.BytesReceived()
		}
		a.pickResult.Done(balancer.DoneInfo{