diff --git a/pkg/remote/trans/common.go b/pkg/remote/trans/common.go index 98a7c4cc2e..93368ddc95 100644 --- a/pkg/remote/trans/common.go +++ b/pkg/remote/trans/common.go @@ -18,6 +18,8 @@ package trans import ( "context" + "errors" + "fmt" "net" "time" @@ -25,7 +27,54 @@ import ( "github.com/cloudwego/kitex/pkg/rpcinfo" ) -var readMoreTimeout = 5 * time.Millisecond +var ( + readMoreTimeout = 5 * time.Millisecond + + ErrRemoteClosed = errors.New("remote connection closed") +) + +// RemoteClosedSource identifies how a remote-closed error was classified. +type RemoteClosedSource int + +const ( + // RemoteClosedByExtension indicates the transport extension recognized the error. + RemoteClosedByExtension RemoteClosedSource = iota + // RemoteClosedByConnectionState indicates the connection was already inactive. + RemoteClosedByConnectionState +) + +// String implements fmt.Stringer for readable logging. +func (s RemoteClosedSource) String() string { + switch s { + case RemoteClosedByExtension: + return "extension" + case RemoteClosedByConnectionState: + return "connection_state" + default: + return "unknown" + } +} + +// RemoteClosedError records the original error and how it was classified. +type RemoteClosedError struct { + Source RemoteClosedSource + Cause error +} + +// Error implements the error interface. +func (e *RemoteClosedError) Error() string { + return fmt.Sprintf("%s (%s): %v", ErrRemoteClosed, e.Source, e.Cause) +} + +// Unwrap exposes the original error. +func (e *RemoteClosedError) Unwrap() error { + return e.Cause +} + +// Is enables errors.Is(err, ErrRemoteClosed). +func (e *RemoteClosedError) Is(target error) bool { + return target == ErrRemoteClosed +} // Extension is the interface that trans extensions need to implement, it will make the extension of trans more easily. // Normally if we want to extend transport layer we need to implement the trans interfaces which are defined in trans_handler.go. @@ -50,6 +99,27 @@ func GetReadTimeout(cfg rpcinfo.RPCConfig) time.Duration { return cfg.RPCTimeout() + readMoreTimeout } +// IsRemoteClosedErr returns the remote-closed classification for err. +// +// Besides the extension's error-based check, it also treats the error as remote-closed +// when conn is already inactive. This covers the case where an encoder flattens the +// underlying netpoll.ErrConnClosed into a plain string (e.g. gopkg/ttheader uses %s), +// which breaks the errors.Is chain so ext.IsRemoteClosedErr can no longer recognize it. +func IsRemoteClosedErr(ext Extension, err error, conn net.Conn) *RemoteClosedError { + if err == nil { + return nil + } + if ext.IsRemoteClosedErr(err) { + return &RemoteClosedError{Source: RemoteClosedByExtension, Cause: err} + } + if ac, ok := conn.(remote.IsActive); ok { + if !ac.IsActive() { + return &RemoteClosedError{Source: RemoteClosedByConnectionState, Cause: err} + } + } + return nil +} + // MuxEnabledFlag is used to determine whether a serverHandlerFactory is multiplexing. type MuxEnabledFlag interface { MuxEnabled() bool diff --git a/pkg/remote/trans/common_test.go b/pkg/remote/trans/common_test.go new file mode 100644 index 0000000000..c5a039bf9e --- /dev/null +++ b/pkg/remote/trans/common_test.go @@ -0,0 +1,81 @@ +/* + * Copyright 2024 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package trans + +import ( + "errors" + "net" + "testing" + + "github.com/cloudwego/kitex/internal/mocks" + "github.com/cloudwego/kitex/internal/test" +) + +// connWithActive is a net.Conn that also reports its active state, +// used to simulate a netpoll-like connection in tests. +type connWithActive struct { + mocks.Conn + active bool +} + +func (c *connWithActive) IsActive() bool { + return c.active +} + +func TestIsRemoteClosedErr(t *testing.T) { + closedByErr := errors.New("remote closed") + otherErr := errors.New("some business error") + + // ext recognizes closedByErr as remote-closed, but nothing else. + ext := &MockExtension{ + IsRemoteClosedErrFunc: func(err error) bool { + return errors.Is(err, closedByErr) + }, + } + + activeConn := &connWithActive{active: true} + inactiveConn := &connWithActive{active: false} + plainConn := &mocks.Conn{} // does not implement remote.IsActive + + cases := []struct { + name string + err error + conn net.Conn + wantErr bool + wantSource RemoteClosedSource + }{ + {name: "nil error", err: nil, conn: inactiveConn}, + {name: "ext recognizes the error", err: closedByErr, conn: activeConn, wantErr: true, wantSource: RemoteClosedByExtension}, + {name: "unrecognized error but conn inactive", err: otherErr, conn: inactiveConn, wantErr: true, wantSource: RemoteClosedByConnectionState}, + {name: "unrecognized error and conn active", err: otherErr, conn: activeConn}, + {name: "unrecognized error and conn without IsActive", err: otherErr, conn: plainConn}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got := IsRemoteClosedErr(ext, c.err, c.conn) + if !c.wantErr { + test.Assert(t, got == nil, c.name, got) + return + } + test.Assert(t, got != nil, c.name) + test.Assert(t, got.Source == c.wantSource, c.name, got.Source, c.wantSource) + test.Assert(t, errors.Is(got, ErrRemoteClosed), c.name, got) + test.Assert(t, errors.Is(got, c.err), c.name, got, c.err) + }) + } +} diff --git a/pkg/remote/trans/default_server_handler.go b/pkg/remote/trans/default_server_handler.go index 0499eef9a6..3f10f25c70 100644 --- a/pkg/remote/trans/default_server_handler.go +++ b/pkg/remote/trans/default_server_handler.go @@ -23,11 +23,13 @@ import ( "net" "runtime/debug" "sync/atomic" + "time" "github.com/cloudwego/kitex/pkg/endpoint" "github.com/cloudwego/kitex/pkg/kerrors" "github.com/cloudwego/kitex/pkg/klog" "github.com/cloudwego/kitex/pkg/remote" + "github.com/cloudwego/kitex/pkg/remote/trans/internal/logbackoff" "github.com/cloudwego/kitex/pkg/rpcinfo" "github.com/cloudwego/kitex/pkg/stats" ) @@ -55,6 +57,7 @@ type svrTransHandler struct { transPipe *remote.TransPipeline ext Extension inGracefulShutdown uint32 + remoteClosedWarn logbackoff.Exponential } // Write implements the remote.ServerTransHandler interface. @@ -159,7 +162,7 @@ func (t *svrTransHandler) OnRead(ctx context.Context, conn net.Conn) (err error) err = panicErr } } - t.finishTracer(ctx, ri, err, panicErr) + t.finishTracer(ctx, ri, err, conn, panicErr) t.finishProfiler(ctx) remote.RecycleMessage(recvMsg) remote.RecycleMessage(sendMsg) @@ -245,8 +248,11 @@ func (t *svrTransHandler) OnInactive(ctx context.Context, conn net.Conn) { func (t *svrTransHandler) OnError(ctx context.Context, err error, conn net.Conn) { ri := rpcinfo.GetRPCInfo(ctx) rService, rAddr := getRemoteInfo(ri, conn) - if t.ext.IsRemoteClosedErr(err) { + if remoteClosedErr := IsRemoteClosedErr(t.ext, err, conn); remoteClosedErr != nil { // it should not regard error which cause by remote connection closed as server error + if count, elapsed, allow := t.remoteClosedWarn.Observe(time.Now()); allow { + klog.CtxWarnf(ctx, "KITEX: processing request error caused by remote connection closed, source=%s, remoteService=%s, remoteAddr=%v, error=%s, count=%d, period=%s", remoteClosedErr.Source, rService, rAddr, remoteClosedErr.Cause, count, elapsed) + } if ri == nil { return } @@ -313,7 +319,7 @@ func (t *svrTransHandler) startTracer(ctx context.Context, ri rpcinfo.RPCInfo) c return c } -func (t *svrTransHandler) finishTracer(ctx context.Context, ri rpcinfo.RPCInfo, err error, panicErr interface{}) { +func (t *svrTransHandler) finishTracer(ctx context.Context, ri rpcinfo.RPCInfo, err error, conn net.Conn, panicErr interface{}) { rpcStats := rpcinfo.AsMutableRPCStats(ri.Stats()) if rpcStats == nil { return @@ -321,7 +327,8 @@ func (t *svrTransHandler) finishTracer(ctx context.Context, ri rpcinfo.RPCInfo, if panicErr != nil { rpcStats.SetPanicked(panicErr) } - if err != nil && t.ext.IsRemoteClosedErr(err) { + if remoteClosedErr := IsRemoteClosedErr(t.ext, err, conn); remoteClosedErr != nil && + remoteClosedErr.Source == RemoteClosedByExtension { // it should not regard the error which caused by remote connection closed as server error err = nil } diff --git a/pkg/remote/trans/default_server_handler_test.go b/pkg/remote/trans/default_server_handler_test.go index 8f2b5a764d..3a67102c6f 100644 --- a/pkg/remote/trans/default_server_handler_test.go +++ b/pkg/remote/trans/default_server_handler_test.go @@ -256,6 +256,66 @@ func TestSvrTransHandlerReadPanic(t *testing.T) { test.Assert(t, strings.Contains(err.Error(), "panic")) } +func TestSvrTransHandlerOnErrorClosedConn(t *testing.T) { + ri := newMockRPCInfo() + ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri) + handler, err := NewDefaultSvrTransHandler(&remote.ServerOption{}, &MockExtension{}) + test.Assert(t, err == nil, err) + + handler.OnError(ctx, errors.New("encode failed"), &connWithActive{}) + + tag, ok := ri.From().Tag(rpcinfo.RemoteClosedTag) + test.Assert(t, ok) + test.Assert(t, tag == "1", tag) +} + +func TestSvrTransHandlerFinishTracerByConnectionState(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + flattenedErr := errors.New("flattened encode error") + mockTracer := stats.NewMockTracer(ctrl) + mockTracer.EXPECT().Finish(gomock.Any()).Do(func(ctx context.Context) { + test.Assert(t, errors.Is(rpcinfo.GetRPCInfo(ctx).Stats().Error(), flattenedErr)) + }) + tracerCtl := &rpcinfo.TraceController{} + tracerCtl.Append(mockTracer) + + rawHandler, err := NewDefaultSvrTransHandler(&remote.ServerOption{TracerCtl: tracerCtl}, &MockExtension{}) + test.Assert(t, err == nil, err) + handler := rawHandler.(*svrTransHandler) + ri := newMockRPCInfo() + ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri) + + handler.finishTracer(ctx, ri, flattenedErr, &connWithActive{}, nil) +} + +func TestSvrTransHandlerFinishTracerByExtension(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + remoteClosedErr := errors.New("remote closed") + mockTracer := stats.NewMockTracer(ctrl) + mockTracer.EXPECT().Finish(gomock.Any()).Do(func(ctx context.Context) { + test.Assert(t, rpcinfo.GetRPCInfo(ctx).Stats().Error() == nil) + }) + tracerCtl := &rpcinfo.TraceController{} + tracerCtl.Append(mockTracer) + + rawHandler, err := NewDefaultSvrTransHandler( + &remote.ServerOption{TracerCtl: tracerCtl}, + &MockExtension{IsRemoteClosedErrFunc: func(err error) bool { + return errors.Is(err, remoteClosedErr) + }}, + ) + test.Assert(t, err == nil, err) + handler := rawHandler.(*svrTransHandler) + ri := newMockRPCInfo() + ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri) + + handler.finishTracer(ctx, ri, remoteClosedErr, &connWithActive{active: true}, nil) +} + func TestSvrTransHandlerOnReadHeartbeat(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() diff --git a/pkg/remote/trans/internal/logbackoff/exponential.go b/pkg/remote/trans/internal/logbackoff/exponential.go new file mode 100644 index 0000000000..d67a82f863 --- /dev/null +++ b/pkg/remote/trans/internal/logbackoff/exponential.go @@ -0,0 +1,67 @@ +/* + * Copyright 2026 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package logbackoff + +import ( + "sync" + "time" +) + +const ( + initialInterval = time.Second + maxInterval = time.Minute + resetAfter = 5 * time.Minute +) + +// Exponential limits repeated logs with exponential backoff. +type Exponential struct { + mu sync.Mutex + lastLog time.Time + interval time.Duration + count uint64 +} + +// Observe records an event and reports its count and duration since the +// previous emitted log when the current event should be logged. +func (b *Exponential) Observe(now time.Time) (count uint64, elapsed time.Duration, allow bool) { + b.mu.Lock() + defer b.mu.Unlock() + + b.count++ + if b.lastLog.IsZero() { + b.lastLog = now + b.interval = initialInterval + count = b.count + b.count = 0 + return count, 0, true + } + elapsed = now.Sub(b.lastLog) + if elapsed < resetAfter && now.Before(b.lastLog.Add(b.interval)) { + return 0, 0, false + } + if elapsed >= resetAfter { + b.interval = initialInterval + } else if b.interval < maxInterval/2 { + b.interval *= 2 + } else { + b.interval = maxInterval + } + b.lastLog = now + count = b.count + b.count = 0 + return count, elapsed, true +} diff --git a/pkg/remote/trans/internal/logbackoff/exponential_test.go b/pkg/remote/trans/internal/logbackoff/exponential_test.go new file mode 100644 index 0000000000..b33db0a6a6 --- /dev/null +++ b/pkg/remote/trans/internal/logbackoff/exponential_test.go @@ -0,0 +1,84 @@ +/* + * Copyright 2026 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package logbackoff + +import ( + "testing" + "time" + + "github.com/cloudwego/kitex/internal/test" +) + +func TestExponentialObserve(t *testing.T) { + var backoff Exponential + now := time.Unix(0, 0) + + count, elapsed, allow := backoff.Observe(now) + test.Assert(t, allow) + test.Assert(t, count == 1, count) + test.Assert(t, elapsed == 0, elapsed) + + count, elapsed, allow = backoff.Observe(now.Add(initialInterval - time.Nanosecond)) + test.Assert(t, !allow) + test.Assert(t, count == 0, count) + test.Assert(t, elapsed == 0, elapsed) + + now = now.Add(initialInterval) + count, elapsed, allow = backoff.Observe(now) + test.Assert(t, allow) + test.Assert(t, count == 2, count) + test.Assert(t, elapsed == initialInterval, elapsed) + + count, elapsed, allow = backoff.Observe(now.Add(2*time.Second - time.Nanosecond)) + test.Assert(t, !allow) + test.Assert(t, count == 0, count) + test.Assert(t, elapsed == 0, elapsed) + + now = now.Add(2 * time.Second) + count, elapsed, allow = backoff.Observe(now) + test.Assert(t, allow) + test.Assert(t, count == 2, count) + test.Assert(t, elapsed == 2*time.Second, elapsed) + for interval := 4 * time.Second; interval <= maxInterval; interval *= 2 { + now = now.Add(interval) + count, elapsed, allow = backoff.Observe(now) + test.Assert(t, allow) + test.Assert(t, count == 1, count) + test.Assert(t, elapsed == interval, elapsed) + } + + count, elapsed, allow = backoff.Observe(now.Add(maxInterval - time.Nanosecond)) + test.Assert(t, !allow) + test.Assert(t, count == 0, count) + test.Assert(t, elapsed == 0, elapsed) + now = now.Add(maxInterval) + count, elapsed, allow = backoff.Observe(now) + test.Assert(t, allow) + test.Assert(t, count == 2, count) + test.Assert(t, elapsed == maxInterval, elapsed) + + now = now.Add(resetAfter) + count, elapsed, allow = backoff.Observe(now) + test.Assert(t, allow) + test.Assert(t, count == 1, count) + test.Assert(t, elapsed == resetAfter, elapsed) + + count, elapsed, allow = backoff.Observe(now.Add(initialInterval - time.Nanosecond)) + test.Assert(t, !allow) + test.Assert(t, count == 0, count) + test.Assert(t, elapsed == 0, elapsed) +} diff --git a/pkg/remote/trans/netpollmux/server_handler.go b/pkg/remote/trans/netpollmux/server_handler.go index fdad9cccee..13e6f497f7 100644 --- a/pkg/remote/trans/netpollmux/server_handler.go +++ b/pkg/remote/trans/netpollmux/server_handler.go @@ -33,6 +33,7 @@ import ( "github.com/cloudwego/kitex/pkg/klog" "github.com/cloudwego/kitex/pkg/remote" "github.com/cloudwego/kitex/pkg/remote/trans" + "github.com/cloudwego/kitex/pkg/remote/trans/internal/logbackoff" np "github.com/cloudwego/kitex/pkg/remote/trans/netpoll" "github.com/cloudwego/kitex/pkg/remote/transmeta" "github.com/cloudwego/kitex/pkg/rpcinfo" @@ -82,15 +83,16 @@ func newSvrTransHandler(opt *remote.ServerOption) (*svrTransHandler, error) { var _ remote.ServerTransHandler = &svrTransHandler{} type svrTransHandler struct { - opt *remote.ServerOption - svcSearcher remote.ServiceSearcher - inkHdlFunc endpoint.Endpoint - codec remote.Codec - transPipe *remote.TransPipeline - ext trans.Extension - funcPool sync.Pool - conns sync.Map - tasks sync.WaitGroup + opt *remote.ServerOption + svcSearcher remote.ServiceSearcher + inkHdlFunc endpoint.Endpoint + codec remote.Codec + transPipe *remote.TransPipeline + ext trans.Extension + funcPool sync.Pool + conns sync.Map + tasks sync.WaitGroup + remoteClosedWarn logbackoff.Exponential } // Write implements the remote.ServerTransHandler interface. @@ -218,10 +220,10 @@ func (t *svrTransHandler) task(muxSvrConnCtx context.Context, conn net.Conn, rea klog.Errorf("KITEX: panic happened, error=%s\nstack=%s", panicErr, string(debug.Stack())) } } + t.finishTracer(ctx, rpcInfo, err, muxSvrConn, panicErr) if closeConn && conn != nil { conn.Close() } - t.finishTracer(ctx, rpcInfo, err, panicErr) remote.RecycleMessage(recvMsg) remote.RecycleMessage(sendMsg) // reset rpcinfo for reuse @@ -385,8 +387,11 @@ func (t *svrTransHandler) GracefulShutdown(ctx context.Context) error { func (t *svrTransHandler) OnError(ctx context.Context, err error, conn net.Conn) { ri := rpcinfo.GetRPCInfo(ctx) rService, rAddr := getRemoteInfo(ri, conn) - if t.ext.IsRemoteClosedErr(err) { + if remoteClosedErr := trans.IsRemoteClosedErr(t.ext, err, conn); remoteClosedErr != nil { // it should not regard error which cause by remote connection closed as server error + if count, elapsed, allow := t.remoteClosedWarn.Observe(time.Now()); allow { + klog.CtxWarnf(ctx, "KITEX: processing request error caused by remote connection closed, source=%s, remoteService=%s, remoteAddr=%v, error=%s, count=%d, period=%s", remoteClosedErr.Source, rService, rAddr, remoteClosedErr.Cause, count, elapsed) + } if ri == nil { return } @@ -457,7 +462,7 @@ func (t *svrTransHandler) startTracer(ctx context.Context, ri rpcinfo.RPCInfo) c return c } -func (t *svrTransHandler) finishTracer(ctx context.Context, ri rpcinfo.RPCInfo, err error, panicErr interface{}) { +func (t *svrTransHandler) finishTracer(ctx context.Context, ri rpcinfo.RPCInfo, err error, conn net.Conn, panicErr interface{}) { rpcStats := rpcinfo.AsMutableRPCStats(ri.Stats()) if rpcStats == nil { return @@ -465,7 +470,8 @@ func (t *svrTransHandler) finishTracer(ctx context.Context, ri rpcinfo.RPCInfo, if panicErr != nil { rpcStats.SetPanicked(panicErr) } - if errors.Is(err, netpoll.ErrConnClosed) { + if remoteClosedErr := trans.IsRemoteClosedErr(t.ext, err, conn); remoteClosedErr != nil && + remoteClosedErr.Source == trans.RemoteClosedByExtension { // it should not regard error which cause by remote connection closed as server error err = nil } diff --git a/pkg/remote/trans/netpollmux/server_handler_test.go b/pkg/remote/trans/netpollmux/server_handler_test.go index e1234af6ba..16972e9d69 100644 --- a/pkg/remote/trans/netpollmux/server_handler_test.go +++ b/pkg/remote/trans/netpollmux/server_handler_test.go @@ -26,10 +26,12 @@ import ( "time" "github.com/cloudwego/netpoll" + "github.com/golang/mock/gomock" "github.com/cloudwego/kitex/internal/mocks" mockmessage "github.com/cloudwego/kitex/internal/mocks/message" mocksremote "github.com/cloudwego/kitex/internal/mocks/remote" + mockstats "github.com/cloudwego/kitex/internal/mocks/stats" "github.com/cloudwego/kitex/internal/test" "github.com/cloudwego/kitex/pkg/remote" "github.com/cloudwego/kitex/pkg/remote/codec" @@ -537,7 +539,138 @@ func TestOnError(t *testing.T) { ctx = rpcinfo.NewCtxWithRPCInfo(ctx, rpcInfo) svrTransHdlr.OnError(ctx, errors.New("test mock err"), conn) + tag, ok := rpcInfo.From().Tag(rpcinfo.RemoteClosedTag) + test.Assert(t, ok) + test.Assert(t, tag == "1", tag) svrTransHdlr.OnError(ctx, netpoll.ErrConnClosed, conn) + + activeConn := &MockNetpollConn{ + Conn: mocks.Conn{ + RemoteAddrFunc: func() net.Addr { + return addr + }, + }, + IsActiveFunc: func() bool { + return true + }, + } + activeRPCInfo := newTestRpcInfo() + activeCtx := rpcinfo.NewCtxWithRPCInfo(context.Background(), activeRPCInfo) + svrTransHdlr.OnError(activeCtx, errors.New("test mock err"), activeConn) + _, ok = activeRPCInfo.From().Tag(rpcinfo.RemoteClosedTag) + test.Assert(t, !ok) +} + +func TestFinishTracerByConnectionState(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + flattenedErr := errors.New("flattened encode error") + mockTracer := mockstats.NewMockTracer(ctrl) + mockTracer.EXPECT().Finish(gomock.Any()).Do(func(ctx context.Context) { + test.Assert(t, errors.Is(rpcinfo.GetRPCInfo(ctx).Stats().Error(), flattenedErr)) + }) + tracerCtl := &rpcinfo.TraceController{} + tracerCtl.Append(mockTracer) + + handler, err := newSvrTransHandler(&remote.ServerOption{TracerCtl: tracerCtl}) + test.Assert(t, err == nil, err) + rpcInfo := newTestRpcInfo() + ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), rpcInfo) + conn := &MockNetpollConn{} + + handler.finishTracer(ctx, rpcInfo, flattenedErr, conn, nil) +} + +func TestFinishTracerByExtension(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockTracer := mockstats.NewMockTracer(ctrl) + mockTracer.EXPECT().Finish(gomock.Any()).Do(func(ctx context.Context) { + test.Assert(t, rpcinfo.GetRPCInfo(ctx).Stats().Error() == nil) + }) + tracerCtl := &rpcinfo.TraceController{} + tracerCtl.Append(mockTracer) + + handler, err := newSvrTransHandler(&remote.ServerOption{TracerCtl: tracerCtl}) + test.Assert(t, err == nil, err) + rpcInfo := newTestRpcInfo() + ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), rpcInfo) + conn := &MockNetpollConn{ + IsActiveFunc: func() bool { + return true + }, + } + + handler.finishTracer(ctx, rpcInfo, netpoll.ErrConnClosed, conn, nil) +} + +func TestTaskFinishTracerPreservesActiveWriteErrorBeforeClose(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + encodeErr := errors.New("encode write failed") + active := true + mockTracer := mockstats.NewMockTracer(ctrl) + mockTracer.EXPECT().Start(gomock.Any()).DoAndReturn(func(ctx context.Context) context.Context { + return ctx + }) + mockTracer.EXPECT().Finish(gomock.Any()).Do(func(ctx context.Context) { + err := rpcinfo.GetRPCInfo(ctx).Stats().Error() + test.Assert(t, errors.Is(err, encodeErr), err) + test.Assert(t, active) + }) + tracerCtl := &rpcinfo.TraceController{} + tracerCtl.Append(mockTracer) + + handler, err := newSvrTransHandler(&remote.ServerOption{ + Codec: &MockCodec{ + EncodeFunc: func(context.Context, remote.Message, remote.ByteBuffer) error { + return encodeErr + }, + DecodeFunc: func(context.Context, remote.Message, remote.ByteBuffer) error { + return nil + }, + }, + TracerCtl: tracerCtl, + InitOrResetRPCInfoFunc: func(ri rpcinfo.RPCInfo, _ net.Addr) rpcinfo.RPCInfo { + return ri + }, + }) + test.Assert(t, err == nil, err) + handler.SetPipeline(remote.NewTransPipeline(handler)) + handler.SetInvokeHandleFunc(func(context.Context, interface{}, interface{}) error { + return nil + }) + + rpcInfo := newTestRpcInfo() + rpcInfo.Invocation().(rpcinfo.InvocationSetter).SetMethodInfo(svcInfo.MethodInfo(context.Background(), mocks.MockMethod)) + pool := &sync.Pool{ + New: func() interface{} { + return rpcInfo + }, + } + conn := &MockNetpollConn{ + Conn: mocks.Conn{ + CloseFunc: func() error { + active = false + return nil + }, + RemoteAddrFunc: func() net.Addr { + return addr + }, + }, + IsActiveFunc: func() bool { + return active + }, + } + muxSvrConn := newMuxSvrConn(conn, pool) + ctx := context.WithValue(context.Background(), ctxKeyMuxSvrConn{}, muxSvrConn) + + handler.task(ctx, conn, netpoll.NewLinkBuffer(1)) + + test.Assert(t, !active) } // TestInvokeNoMethod test invoke no method