From d5bc85c1e9a36225ecc02757efcc874e13962973 Mon Sep 17 00:00:00 2001
From: MakMukhi <mmukhi@google.com>
Date: Thu, 1 Jun 2017 11:57:45 -0700
Subject: [PATCH] Server shouldn't Fatalf in case it fails to encode. (#1251)

* Server shouldn't Fatalf in case it fails to encode.

* golint

* post-review update
---
 call.go              |  2 +-
 rpc_util.go          |  4 ++--
 server.go            | 10 ++--------
 stream.go            |  3 +--
 test/end2end_test.go | 44 ++++++++++++++++++++++++++++++++++++++++++++
 5 files changed, 50 insertions(+), 13 deletions(-)

diff --git a/call.go b/call.go
index a2b89ac6a36e..e73812fd18b4 100644
--- a/call.go
+++ b/call.go
@@ -119,7 +119,7 @@ func sendRequest(ctx context.Context, dopts dialOptions, compressor Compressor,
 	}
 	outBuf, err := encode(dopts.codec, args, compressor, cbuf, outPayload)
 	if err != nil {
-		return Errorf(codes.Internal, "grpc: %v", err)
+		return err
 	}
 	if c.maxSendMessageSize == nil {
 		return Errorf(codes.Internal, "callInfo maxSendMessageSize field uninitialized(nil)")
diff --git a/rpc_util.go b/rpc_util.go
index a24257afe907..9862b32725ca 100644
--- a/rpc_util.go
+++ b/rpc_util.go
@@ -314,7 +314,7 @@ func encode(c Codec, msg interface{}, cp Compressor, cbuf *bytes.Buffer, outPayl
 		// TODO(zhaoq): optimize to reduce memory alloc and copying.
 		b, err = c.Marshal(msg)
 		if err != nil {
-			return nil, err
+			return nil, Errorf(codes.Internal, "grpc: error while marshaling: %v", err.Error())
 		}
 		if outPayload != nil {
 			outPayload.Payload = msg
@@ -324,7 +324,7 @@ func encode(c Codec, msg interface{}, cp Compressor, cbuf *bytes.Buffer, outPayl
 		}
 		if cp != nil {
 			if err := cp.Do(cbuf, b); err != nil {
-				return nil, err
+				return nil, Errorf(codes.Internal, "grpc: error while compressing: %v", err.Error())
 			}
 			b = cbuf.Bytes()
 		}
diff --git a/server.go b/server.go
index c2b523c7acd3..9e9a3442ef57 100644
--- a/server.go
+++ b/server.go
@@ -664,14 +664,8 @@ func (s *Server) sendResponse(t transport.ServerTransport, stream *transport.Str
 	}
 	p, err := encode(s.opts.codec, msg, cp, cbuf, outPayload)
 	if err != nil {
-		// This typically indicates a fatal issue (e.g., memory
-		// corruption or hardware faults) the application program
-		// cannot handle.
-		//
-		// TODO(zhaoq): There exist other options also such as only closing the
-		// faulty stream locally and remotely (Other streams can keep going). Find
-		// the optimal option.
-		grpclog.Fatalf("grpc: Server failed to encode response %v", err)
+		grpclog.Println("grpc: server failed to encode response: ", err)
+		return err
 	}
 	if len(p) > s.opts.maxSendMessageSize {
 		return status.Errorf(codes.ResourceExhausted, "grpc: trying to send message larger than max (%d vs. %d)", len(p), s.opts.maxSendMessageSize)
diff --git a/stream.go b/stream.go
index ed0ebe7b73ec..2106d3ecb28f 100644
--- a/stream.go
+++ b/stream.go
@@ -364,7 +364,7 @@ func (cs *clientStream) SendMsg(m interface{}) (err error) {
 		}
 	}()
 	if err != nil {
-		return Errorf(codes.Internal, "grpc: %v", err)
+		return err
 	}
 	if cs.c.maxSendMessageSize == nil {
 		return Errorf(codes.Internal, "callInfo maxSendMessageSize field uninitialized(nil)")
@@ -606,7 +606,6 @@ func (ss *serverStream) SendMsg(m interface{}) (err error) {
 		}
 	}()
 	if err != nil {
-		err = Errorf(codes.Internal, "grpc: %v", err)
 		return err
 	}
 	if len(out) > ss.maxSendMessageSize {
diff --git a/test/end2end_test.go b/test/end2end_test.go
index 0456cb7ab25f..141048300dc2 100644
--- a/test/end2end_test.go
+++ b/test/end2end_test.go
@@ -454,6 +454,7 @@ type test struct {
 	clientInitialWindowSize     int32
 	clientInitialConnWindowSize int32
 	perRPCCreds                 credentials.PerRPCCredentials
+	customCodec                 grpc.Codec
 
 	// srv and srvAddr are set once startServer is called.
 	srv     *grpc.Server
@@ -4795,3 +4796,46 @@ func testPerRPCCredentialsViaDialOptionsAndCallOptions(t *testing.T, e env) {
 		t.Fatalf("Test failed. Reason: %v", err)
 	}
 }
+
+type errCodec struct {
+	noError bool
+}
+
+func (c *errCodec) Marshal(v interface{}) ([]byte, error) {
+	if c.noError {
+		return []byte{}, nil
+	}
+	return nil, fmt.Errorf("3987^12 + 4365^12 = 4472^12")
+}
+
+func (c *errCodec) Unmarshal(data []byte, v interface{}) error {
+	return nil
+}
+
+func (c *errCodec) String() string {
+	return "Fermat's near-miss."
+}
+
+func TestEncodeDoesntPanic(t *testing.T) {
+	defer leakCheck(t)()
+	for _, e := range listTestEnv() {
+		testEncodeDoesntPanic(t, e)
+	}
+}
+
+func testEncodeDoesntPanic(t *testing.T, e env) {
+	te := newTest(t, e)
+	erc := &errCodec{}
+	te.customCodec = erc
+	te.startServer(&testServer{security: e.security})
+	defer te.tearDown()
+	te.customCodec = nil
+	tc := testpb.NewTestServiceClient(te.clientConn())
+	// Failure case, should not panic.
+	tc.EmptyCall(context.Background(), &testpb.Empty{})
+	erc.noError = true
+	// Passing case.
+	if _, err := tc.EmptyCall(context.Background(), &testpb.Empty{}); err != nil {
+		t.Fatalf("EmptyCall(_, _) = _, %v, want _, <nil>", err)
+	}
+}