fix(remote): treat encode failure on closed conn as remote-closed in …

pkg/remote/trans/common.go

@@ -18,14 +18,63 @@ package trans
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"net"
 	"time"
 
 	"github.com/cloudwego/kitex/pkg/remote"
 	"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

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)
+		})
+	}
+}

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,15 +319,16 @@ 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
 	}
 	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
 	}

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()

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
+}