fix: fix the issue which the server-side did not report failure after a panic occurred during request processing

Author: YangruiEmma

pkg/remote/trans/default_server_handler.go

@@ -144,10 +144,11 @@ func (t *svrTransHandler) OnRead(ctx context.Context, conn net.Conn) (err error)
 			} else {
 				klog.CtxErrorf(ctx, "KITEX: panic happened, error=%v\nstack=%s", panicErr, stack)
 			}
-			if err != nil {
-				wrapErr = kerrors.ErrPanic.WithCauseAndStack(fmt.Errorf("[happened in OnRead] %s, last error=%s", panicErr, err.Error()), stack)
-			} else {
+			if err == nil {
+				// if err is nil, set err as panic wrapErr
 				wrapErr = kerrors.ErrPanic.WithCauseAndStack(fmt.Errorf("[happened in OnRead] %s", panicErr), stack)
+				err = remote.NewTransError(remote.InternalError, wrapErr)
+				t.writeErrorReplyIfNeeded(ctx, recvMsg, conn, err, ri, true)
 			}
 		}
 		t.finishTracer(ctx, ri, err, panicErr)
@@ -158,10 +159,9 @@ func (t *svrTransHandler) OnRead(ctx context.Context, conn net.Conn) (err error)
 		if rpcinfo.PoolEnabled() {
 			t.opt.InitOrResetRPCInfoFunc(ri, conn.RemoteAddr())
 		}
-		if wrapErr != nil {
-			err = wrapErr
-		}
 		if err != nil && !closeConnOutsideIfErr {
+			// when error is not nil, outside will close conn,
+			// set err not nil just indicate that this kind of error don't need close conn
 			err = nil
 		}
 	}()
@@ -186,7 +186,7 @@ func (t *svrTransHandler) OnRead(ctx context.Context, conn net.Conn) (err error)
 		// reply processing
 		var methodInfo serviceinfo.MethodInfo
 		if methodInfo, err = GetMethodInfo(ri, svcInfo); err != nil {
-			// it won't be err, because the method has been checked in decode, err check here just do defensive inspection
+			// it won't be error, because the method has been checked in decode, err check here just do defensive inspection
 			t.writeErrorReplyIfNeeded(ctx, recvMsg, conn, err, ri, true)
 			// for proxy case, need read actual remoteAddr, error print must exec after writeErrorReplyIfNeeded,
 			// t.OnError(ctx, err, conn) will be executed at outer function when transServer close the conn
@@ -278,6 +278,12 @@ func (t *svrTransHandler) writeErrorReplyIfNeeded(
 		// conn is closed, no need reply
 		return
 	}
+	defer func() {
+		if r := recover(); r != nil {
+			rService, rAddr := getRemoteInfo(ri, conn)
+			klog.CtxErrorf(ctx, "KITEX: write error reply panic, remoteAddress=%s, remoteService=%s, error=%v\nstack=%s", rAddr, rService, r, string(debug.Stack()))
+		}
+	}()
 	svcInfo := recvMsg.ServiceInfo()
 	if svcInfo != nil {
 		if methodInfo, _ := GetMethodInfo(ri, svcInfo); methodInfo != nil {

pkg/remote/trans/default_server_handler_test.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"net"
+	"strings"
 	"testing"
 
 	"github.com/golang/mock/gomock"
@@ -211,6 +212,59 @@ func TestSvrTransHandlerReadErr(t *testing.T) {
 	test.Assert(t, errors.Is(err, mockErr))
 }
 
+func TestSvrTransHandlerReadPanic(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockTracer := stats.NewMockTracer(ctrl)
+	mockTracer.EXPECT().Start(gomock.Any()).DoAndReturn(func(ctx context.Context) context.Context { return ctx }).AnyTimes()
+	mockTracer.EXPECT().Finish(gomock.Any()).DoAndReturn(func(ctx context.Context) {
+		err := rpcinfo.GetRPCInfo(ctx).Stats().Error()
+		test.Assert(t, err != nil)
+	}).AnyTimes()
+
+	buf := remote.NewReaderWriterBuffer(1024)
+	ext := &MockExtension{
+		NewWriteByteBufferFunc: func(ctx context.Context, conn net.Conn, msg remote.Message) remote.ByteBuffer {
+			return buf
+		},
+		NewReadByteBufferFunc: func(ctx context.Context, conn net.Conn, msg remote.Message) remote.ByteBuffer {
+			return buf
+		},
+	}
+
+	tracerCtl := &rpcinfo.TraceController{}
+	tracerCtl.Append(mockTracer)
+	opt := &remote.ServerOption{
+		Codec: &MockCodec{
+			EncodeFunc: func(ctx context.Context, msg remote.Message, out remote.ByteBuffer) error {
+				return nil
+			},
+			DecodeFunc: func(ctx context.Context, msg remote.Message, in remote.ByteBuffer) error {
+				panic("mock")
+			},
+		},
+		SvcSearcher:   svcSearcher,
+		TargetSvcInfo: svcInfo,
+		TracerCtl:     tracerCtl,
+		InitOrResetRPCInfoFunc: func(ri rpcinfo.RPCInfo, addr net.Addr) rpcinfo.RPCInfo {
+			rpcinfo.AsMutableEndpointInfo(ri.From()).SetAddress(addr)
+			return ri
+		},
+	}
+	ri := rpcinfo.NewRPCInfo(rpcinfo.EmptyEndpointInfo(), rpcinfo.FromBasicInfo(&rpcinfo.EndpointBasicInfo{}),
+		rpcinfo.NewInvocation("", ""), nil, rpcinfo.NewRPCStats())
+	ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)
+
+	svrHandler, err := NewDefaultSvrTransHandler(opt, ext)
+	test.Assert(t, err == nil)
+	pl := remote.NewTransPipeline(svrHandler)
+	svrHandler.SetPipeline(pl)
+	err = svrHandler.OnRead(ctx, &mocks.Conn{})
+	test.Assert(t, err != nil)
+	test.Assert(t, strings.Contains(err.Error(), "panic"))
+}
+
 func TestSvrTransHandlerOnReadHeartbeat(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()

pkg/remote/trans/netpoll/server_handler_test.go

@@ -172,8 +172,8 @@ func TestInvokeErr(t *testing.T) {
 	test.Assert(t, isInvoked)
 }
 
-// TestPanicAfterRead test server_handler not panic after read
-func TestPanicAfterRead(t *testing.T) {
+// TestPipelineNilPanic test server_handler that TransPipeline is nil
+func TestPipelineNilPanic(t *testing.T) {
 	// 1. prepare mock data
 	var isWriteBufFlushed bool
 	var isReaderBufReleased bool