Reopening: Server shouldn't Fatalf in case it fails to encode. (#1276)

MakMukhi · web-flow · commit 6fecf2831adf · 2017-06-02T12:32:37.000-07:00
* Server shouldn't Fatalf in case it fails to encode.
diff --git 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
@@ -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
@@ -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
@@ -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
@@ -4795,3 +4795,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)
+	}
+}

Original file line number	Diff line number	Diff line change
`@@ -119,7 +119,7 @@ func sendRequest(ctx context.Context, dopts dialOptions, compressor Compressor,`
`119`	`119`	`}`
`120`	`120`	`outBuf, err := encode(dopts.codec, args, compressor, cbuf, outPayload)`
`121`	`121`	`if err != nil {`
`122`		`- return Errorf(codes.Internal, "grpc: %v", err)`
	`122`	`+ return err`
`123`	`123`	`}`
`124`	`124`	`if c.maxSendMessageSize == nil {`
`125`	`125`	`return Errorf(codes.Internal, "callInfo maxSendMessageSize field uninitialized(nil)")`
Original file line number	Diff line number	Diff line change
`@@ -314,7 +314,7 @@ func encode(c Codec, msg interface{}, cp Compressor, cbuf *bytes.Buffer, outPayl`
`314`	`314`	`// TODO(zhaoq): optimize to reduce memory alloc and copying.`
`315`	`315`	`b, err = c.Marshal(msg)`
`316`	`316`	`if err != nil {`
`317`		`- return nil, err`
	`317`	`+ return nil, Errorf(codes.Internal, "grpc: error while marshaling: %v", err.Error())`
`318`	`318`	`}`
`319`	`319`	`if outPayload != nil {`
`320`	`320`	`outPayload.Payload = msg`
`@@ -324,7 +324,7 @@ func encode(c Codec, msg interface{}, cp Compressor, cbuf *bytes.Buffer, outPayl`
`324`	`324`	`}`
`325`	`325`	`if cp != nil {`
`326`	`326`	`if err := cp.Do(cbuf, b); err != nil {`
`327`		`- return nil, err`
	`327`	`+ return nil, Errorf(codes.Internal, "grpc: error while compressing: %v", err.Error())`
`328`	`328`	`}`
`329`	`329`	`b = cbuf.Bytes()`
`330`	`330`	`}`
Original file line number	Diff line number	Diff line change
`@@ -364,7 +364,7 @@ func (cs *clientStream) SendMsg(m interface{}) (err error) {`
`364`	`364`	`}`
`365`	`365`	`}()`
`366`	`366`	`if err != nil {`
`367`		`- return Errorf(codes.Internal, "grpc: %v", err)`
	`367`	`+ return err`
`368`	`368`	`}`
`369`	`369`	`if cs.c.maxSendMessageSize == nil {`
`370`	`370`	`return Errorf(codes.Internal, "callInfo maxSendMessageSize field uninitialized(nil)")`
`@@ -606,7 +606,6 @@ func (ss *serverStream) SendMsg(m interface{}) (err error) {`
`606`	`606`	`}`
`607`	`607`	`}()`
`608`	`608`	`if err != nil {`
`609`		`- err = Errorf(codes.Internal, "grpc: %v", err)`
`610`	`609`	`return err`
`611`	`610`	`}`
`612`	`611`	`if len(out) > ss.maxSendMessageSize {`