Repository: cloudwego/hertz
Pull request #1322 (2025-04-07 12:57:14)

feat: std transport sense client connection close
#### What type of PR is this?
feat
<!--
Add one of the following kinds:

build: Changes that affect the build system or external dependencies (example scopes: gulp, broccoli, npm)
ci: Changes to our CI configuration files and scripts (example scopes: Travis, Circle, BrowserStack, SauceLabs)
docs: Documentation only changes
feat: A new feature
optimize: A new optimization
fix: A bug fix
perf: A code change that improves performance
refactor: A code change that neither fixes a bug nor adds a feature
style: Changes that do not affect the meaning of the code (white space, formatting, missing semi-colons, etc)
test: Adding missing tests or correcting existing tests
chore: Changes to the build process or auxiliary tools and libraries such as documentation generation
-->

#### Check the PR title.
<!--
The description of the title will be attached in Release Notes, 
so please describe it from user-oriented, what this PR does / why we need it.
Please check your PR title with the below requirements:
-->
- [ ] This PR title match the format: \<type\>(optional scope): \<description\>
- [ ] The description of this PR title is user-oriented and clear enough for others to understand.
- [ ] Attach the PR updating the user documentation if the current PR requires user awareness at the usage level. [User docs repo](https://github.com/cloudwego/cloudwego.github.io)


#### (Optional) Translate the PR title into Chinese.


#### (Optional) More detailed description for this PR(en: English/zh: Chinese).
<!--
Provide more detailed info for review(e.g., it's recommended to provide perf data if this is a perf type PR).
-->
en:
zh(optional):

#### (Optional) Which issue(s) this PR fixes:
<!--
Automatically closes linked issue when PR is merged.
Eg: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->

#### (Optional) The PR that updates user documentation:
<!--
If the current PR requires user awareness at the usage level, please submit a PR to update user docs. [User docs repo](https://github.com/cloudwego/cloudwego.github.io)
-->

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/cloudwego/hertz/pull/1322?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) Report
Attention: Patch coverage is `20.51282%` with `62 lines` in your changes missing coverage. Please review.
> Project coverage is 72.51%. Comparing base [(`907afa3`)](https://app.codecov.io/gh/cloudwego/hertz/commit/907afa3bb9d4723a87c59f6ed6ca1f437591daab?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) to head [(`50e5113`)](https://app.codecov.io/gh/cloudwego/hertz/commit/50e5113eff6d25a3385228bc90112b3746c192ea?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).
> Report is 2 commits behind head on develop.

| [Files with missing lines](https://app.codecov.io/gh/cloudwego/hertz/pull/1322?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) | Patch % | Lines |
|---|---|---|
| [pkg/network/standard/connection.go](https://app.codecov.io/gh/cloudwego/hertz/pull/1322?src=pr&el=tree&filepath=pkg%2Fnetwork%2Fstandard%2Fconnection.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL25ldHdvcmsvc3RhbmRhcmQvY29ubmVjdGlvbi5nbw==) | 0.00% | [44 Missing :warning: ](https://app.codecov.io/gh/cloudwego/hertz/pull/1322?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) |
| [pkg/protocol/http1/server.go](https://app.codecov.io/gh/cloudwego/hertz/pull/1322?src=pr&el=tree&filepath=pkg%2Fprotocol%2Fhttp1%2Fserver.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL2h0dHAxL3NlcnZlci5nbw==) | 15.78% | [12 Missing and 4 partials :warning: ](https://app.codecov.io/gh/cloudwego/hertz/pull/1322?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) |
| [pkg/network/standard/transport.go](https://app.codecov.io/gh/cloudwego/hertz/pull/1322?src=pr&el=tree&filepath=pkg%2Fnetwork%2Fstandard%2Ftransport.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL25ldHdvcmsvc3RhbmRhcmQvdHJhbnNwb3J0Lmdv) | 86.66% | [2 Missing :warning: ](https://app.codecov.io/gh/cloudwego/hertz/pull/1322?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) |

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##           develop    #1322      +/-   ##
===========================================
- Coverage    72.78%   72.51%   -0.27%     
===========================================
  Files          139      139              
  Lines        16244    16313      +69     
===========================================
+ Hits         11823    11830       +7     
- Misses        3824     3882      +58     
- Partials       597      601       +4     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/cloudwego/hertz/pull/1322?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>

Patch:
diff --git a/.gitignore b/.gitignore
index d4c5801..c84307a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,6 +3,8 @@
 pkg/app/fs.go.hertz.gz
 coverage.txt
 coverage.out
+# OSX trash
+.DS_Store
 
 # test benchmark tmp output
 cpu.out
diff --git a/internal/network/standard.go b/internal/network/standard.go
new file mode 100644
index 0000000..8e444d1
--- /dev/null
+++ b/internal/network/standard.go
@@ -0,0 +1,25 @@
+// Copyright 2025 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 network
+
+import (
+	"github.com/cloudwego/hertz/pkg/network"
+)
+
+type StatefulConn interface {
+	network.Conn
+	DetectConnectionClose()
+	AbortBlockingRead()
+}
diff --git a/pkg/network/standard/connection.go b/pkg/network/standard/connection.go
index 13c10b4..de1ab98 100644
--- a/pkg/network/standard/connection.go
+++ b/pkg/network/standard/connection.go
@@ -17,6 +17,7 @@
 package standard
 
 import (
+	"context"
 	"crypto/tls"
 	"errors"
 	"io"
@@ -26,6 +27,7 @@ import (
 	"syscall"
 	"time"
 
+	internalNetwork "github.com/cloudwego/hertz/internal/network"
 	errs "github.com/cloudwego/hertz/pkg/common/errors"
 	"github.com/cloudwego/hertz/pkg/common/hlog"
 	"github.com/cloudwego/hertz/pkg/network"
@@ -638,3 +640,55 @@ func newTLSConn(c net.Conn, size int) network.Conn {
 		},
 	}
 }
+
+var earliestTime = time.Unix(1, 0)
+
+var _ internalNetwork.StatefulConn = &statefulConn{}
+
+func NewStatefulConn(c network.Conn, onReadError func(err error)) internalNetwork.StatefulConn {
+	return &statefulConn{
+		Conn:        c,
+		onReadError: onReadError,
+		ch:          make(chan struct{}, 1),
+	}
+}
+
+type statefulConn struct {
+	network.Conn
+	onReadError func(err error)
+	ch          chan struct{}
+}
+
+// DetectConnectionClose starts a blocking read to detect connection error
+func (c *statefulConn) DetectConnectionClose() {
+	c.Conn.SetReadDeadline(time.Time{}) // blocking read
+	go func() {
+		// start a blocking read, if return
+		// 1. n != 0, err == nil: pipeline request
+		// 2. err != nil: execute onReadError
+		_, err := c.Conn.Peek(1) // peek 1 byte to trigger Read Syscall
+		if err != nil {
+			c.onReadError(err)
+		}
+		c.ch <- struct{}{}
+	}()
+}
+
+// AbortBlockingRead cancels the blocking read and waits for the DetectConnectionClose exit.
+// Only call AbortBlockingRead if already called DetectConnectionClose.
+func (c *statefulConn) AbortBlockingRead() {
+	c.Conn.SetReadDeadline(earliestTime) // cancel the blocking read
+	<-c.ch                               // wait until the blocking read returns
+}
+
+func newOnReadErrorFunc(cancelCtx context.CancelFunc) func(err error) {
+	return func(err error) {
+		// 1. timeout error, triggered by `abortBlockingRead` when response is written, ignore.
+		// 2. other connection error (e.g. EOF), cancel the context
+		if ne, ok := err.(net.Error); ok && ne.Timeout() {
+			// ignore the error triggered by AbortBlockingRead, which should be "i/o timeout"
+		} else {
+			cancelCtx()
+		}
+	}
+}
diff --git a/pkg/network/standard/transport.go b/pkg/network/standard/transport.go
index 10ee991..1b08509 100644
--- a/pkg/network/standard/transport.go
+++ b/pkg/network/standard/transport.go
@@ -38,16 +38,17 @@ type transport struct {
 	// and/or multi-KB headers (for example, BIG cookies).
 	//
 	// Default buffer size is used if not set.
-	readBufferSize   int
-	network          string
-	addr             string
-	keepAliveTimeout time.Duration
-	readTimeout      time.Duration
-	handler          network.OnData
-	tls              *tls.Config
-	listenConfig     *net.ListenConfig
-	OnAccept         func(conn net.Conn) context.Context
-	OnConnect        func(ctx context.Context, conn network.Conn) context.Context
+	readBufferSize           int
+	network                  string
+	addr                     string
+	keepAliveTimeout         time.Duration
+	senseClientDisconnection bool
+	readTimeout              time.Duration
+	handler                  network.OnData
+	tls                      *tls.Config
+	listenConfig             *net.ListenConfig
+	OnAccept                 func(conn net.Conn) context.Context
+	OnConnect                func(ctx context.Context, conn network.Conn) context.Context
 
 	// active connections. it +1 after accept and -1 after handler returns
 	active int32
@@ -101,8 +102,16 @@ func (t *transport) serve() (err error) {
 		if t.OnConnect != nil {
 			ctx = t.OnConnect(ctx, c)
 		}
+
 		go func(ctx context.Context, conn network.Conn) {
-			t.handler(ctx, conn)
+			if t.senseClientDisconnection {
+				ctx, cancelCtx := context.WithCancel(ctx)
+				conn = NewStatefulConn(conn, newOnReadErrorFunc(cancelCtx))
+				t.handler(ctx, conn)
+				cancelCtx()
+			} else {
+				t.handler(ctx, conn)
+			}
 			t.updateActive(-1)
 		}(ctx, c)
 	}
@@ -169,14 +178,15 @@ func (t *transport) Shutdown(ctx context.Context) error {
 // For transporter switch
 func NewTransporter(options *config.Options) network.Transporter {
 	return &transport{
-		readBufferSize:   options.ReadBufferSize,
-		network:          options.Network,
-		addr:             options.Addr,
-		keepAliveTimeout: options.KeepAliveTimeout,
-		readTimeout:      options.ReadTimeout,
-		tls:              options.TLS,
-		listenConfig:     options.ListenConfig,
-		OnAccept:         options.OnAccept,
-		OnConnect:        options.OnConnect,
+		readBufferSize:           options.ReadBufferSize,
+		network:                  options.Network,
+		addr:                     options.Addr,
+		keepAliveTimeout:         options.KeepAliveTimeout,
+		readTimeout:              options.ReadTimeout,
+		senseClientDisconnection: options.SenseClientDisconnection,
+		tls:                      options.TLS,
+		listenConfig:             options.ListenConfig,
+		OnAccept:                 options.OnAccept,
+		OnConnect:                options.OnConnect,
 	}
 }
diff --git a/pkg/protocol/http1/server.go b/pkg/protocol/http1/server.go
index eb46f65..c4b86de 100644
--- a/pkg/protocol/http1/server.go
+++ b/pkg/protocol/http1/server.go
@@ -26,6 +26,7 @@ import (
 	"time"
 
 	"github.com/cloudwego/hertz/internal/bytestr"
+	internalNetwork "github.com/cloudwego/hertz/internal/network"
 	internalStats "github.com/cloudwego/hertz/internal/stats"
 	"github.com/cloudwego/hertz/pkg/app"
 	"github.com/cloudwego/hertz/pkg/app/server/render"
@@ -132,6 +133,10 @@ func (s Server) Serve(c context.Context, conn network.Conn) (err error) {
 		cc = c
 	)
 
+	// for sensing connection close
+	// only if `conn` is internalNetwork.StatefulConn
+	statefulConn, configuredSenseConnClose := conn.(internalNetwork.StatefulConn)
+
 	if s.EnableTrace {
 		eventsToTrigger = s.eventStackPool.Get().(*eventStack)
 	}
@@ -167,6 +172,7 @@ func (s Server) Serve(c context.Context, conn network.Conn) (err error) {
 
 	ctx.HTMLRender = s.HTMLRender
 	ctx.SetConn(conn)
+
 	ctx.Request.SetIsTLS(s.TLS != nil)
 	ctx.SetEnableTrace(s.EnableTrace)
 
@@ -178,6 +184,7 @@ func (s Server) Serve(c context.Context, conn network.Conn) (err error) {
 
 	for {
 		connRequestNum++
+		senseConnClose := configuredSenseConnClose // use to check if senseConnClose logic should be triggered for each request
 
 		if zr == nil {
 			zr = ctx.GetReader()
@@ -294,6 +301,7 @@ func (s Server) Serve(c context.Context, conn network.Conn) (err error) {
 				} else {
 					err = req.ContinueReadBody(&ctx.Request, zr, s.MaxRequestBodySize, !s.DisablePreParseMultipartForm)
 				}
+
 				if err != nil {
 					writeErrorResponse(zw, ctx, serverName, err)
 					return
@@ -313,6 +321,13 @@ func (s Server) Serve(c context.Context, conn network.Conn) (err error) {
 				internalStats.Record(ti, stats.ServerHandleFinish, err)
 			})
 		}
+
+		// If request sets BodyStream, we don't start connection close detection logic to prevent concurrent Read.
+		senseConnClose = senseConnClose && !ctx.Request.IsBodyStream()
+		if senseConnClose {
+			statefulConn.DetectConnectionClose()
+		}
+
 		// Handle the request
 		//
 		// NOTE: All middlewares and business handler will be executed in this. And at this point, the request has been parsed
@@ -365,6 +380,13 @@ func (s Server) Serve(c context.Context, conn network.Conn) (err error) {
 			}
 		}
 
+		// Abort the blocking read in DetectConnectionClose
+		if senseConnClose {
+			// this should be sync to wait the blocking read return.
+			// we should make sure the read is not affected by other SetReadDeadline.
+			statefulConn.AbortBlockingRead()
+		}
+
 		// Release the zeroCopyReader before flush to prevent data race
 		if zr != nil {
 			zr.Release() //nolint:errcheck
@@ -407,6 +429,7 @@ func (s Server) Serve(c context.Context, conn network.Conn) (err error) {
 		}
 		// Back to network layer to trigger.
 		// For now, only netpoll network mode has this feature.
+		// FIXME: check
 		if s.IdleTimeout == 0 {
 			return
 		}


Test patch:
diff --git a/pkg/app/server/hertz_test.go b/pkg/app/server/hertz_test.go
index 2a71947..e1aadb3 100644
--- a/pkg/app/server/hertz_test.go
+++ b/pkg/app/server/hertz_test.go
@@ -1234,3 +1234,59 @@ func TestWithDisableDefaultContentType(t *testing.T) {
 	r, _ := hc.Get(testutils.GetURL(h, "")) //nolint:errcheck
 	assert.DeepEqual(t, "", r.Header.Get("Content-Type"))
 }
+
+func TestWithSenseClientDisconnection(t *testing.T) {
+	var closeFlag int32
+	h := New(WithHostPorts("127.0.0.1:0"), WithSenseClientDisconnection(true))
+	h.GET("/ping", func(c context.Context, ctx *app.RequestContext) {
+		assert.DeepEqual(t, "aa", string(ctx.Host()))
+		ch := make(chan struct{})
+		select {
+		case <-c.Done():
+			atomic.StoreInt32(&closeFlag, 1)
+			assert.DeepEqual(t, context.Canceled, c.Err())
+		case <-ch:
+		}
+	})
+	go h.Spin()
+	waitEngineRunning(h)
+
+	con, err := net.Dial("tcp", testutils.GetListenerAddr(h))
+	assert.Nil(t, err)
+	_, err = con.Write([]byte("GET /ping HTTP/1.1\r\nHost: aa\r\n\r\n"))
+	assert.Nil(t, err)
+	time.Sleep(20 * time.Millisecond)
+	assert.DeepEqual(t, atomic.LoadInt32(&closeFlag), int32(0))
+	assert.Nil(t, con.Close())
+	time.Sleep(20 * time.Millisecond)
+	assert.DeepEqual(t, atomic.LoadInt32(&closeFlag), int32(1))
+}
+
+func TestWithSenseClientDisconnectionAndWithOnConnect(t *testing.T) {
+	var closeFlag int32
+	h := New(WithHostPorts("127.0.0.1:0"), WithSenseClientDisconnection(true), WithOnConnect(func(ctx context.Context, conn network.Conn) context.Context {
+		return ctx
+	}))
+	h.GET("/ping", func(c context.Context, ctx *app.RequestContext) {
+		assert.DeepEqual(t, "aa", string(ctx.Host()))
+		ch := make(chan struct{})
+		select {
+		case <-c.Done():
+			atomic.StoreInt32(&closeFlag, 1)
+			assert.DeepEqual(t, context.Canceled, c.Err())
+		case <-ch:
+		}
+	})
+	go h.Spin()
+	waitEngineRunning(h)
+
+	con, err := net.Dial("tcp", testutils.GetListenerAddr(h))
+	assert.Nil(t, err)
+	_, err = con.Write([]byte("GET /ping HTTP/1.1\r\nHost: aa\r\n\r\n"))
+	assert.Nil(t, err)
+	time.Sleep(20 * time.Millisecond)
+	assert.DeepEqual(t, atomic.LoadInt32(&closeFlag), int32(0))
+	assert.Nil(t, con.Close())
+	time.Sleep(20 * time.Millisecond)
+	assert.DeepEqual(t, atomic.LoadInt32(&closeFlag), int32(1))
+}
diff --git a/pkg/app/server/hertz_unix_test.go b/pkg/app/server/hertz_unix_test.go
index 7e14c94..cad021a 100644
--- a/pkg/app/server/hertz_unix_test.go
+++ b/pkg/app/server/hertz_unix_test.go
@@ -35,7 +35,6 @@ import (
 	c "github.com/cloudwego/hertz/pkg/app/client"
 	"github.com/cloudwego/hertz/pkg/common/test/assert"
 	"github.com/cloudwego/hertz/pkg/common/utils"
-	"github.com/cloudwego/hertz/pkg/network"
 	"github.com/cloudwego/hertz/pkg/network/standard"
 	"github.com/cloudwego/hertz/pkg/protocol/consts"
 	"golang.org/x/sys/unix"
@@ -139,59 +138,3 @@ func TestHertz_Spin(t *testing.T) {
 	<-ch2
 	assert.DeepEqual(t, uint32(1), atomic.LoadUint32(&testint))
 }
-
-func TestWithSenseClientDisconnection(t *testing.T) {
-	var closeFlag int32
-	h := New(WithHostPorts("127.0.0.1:0"), WithSenseClientDisconnection(true))
-	h.GET("/ping", func(c context.Context, ctx *app.RequestContext) {
-		assert.DeepEqual(t, "aa", string(ctx.Host()))
-		ch := make(chan struct{})
-		select {
-		case <-c.Done():
-			atomic.StoreInt32(&closeFlag, 1)
-			assert.DeepEqual(t, context.Canceled, c.Err())
-		case <-ch:
-		}
-	})
-	go h.Spin()
-	waitEngineRunning(h)
-
-	con, err := net.Dial("tcp", testutils.GetListenerAddr(h))
-	assert.Nil(t, err)
-	_, err = con.Write([]byte("GET /ping HTTP/1.1\r\nHost: aa\r\n\r\n"))
-	assert.Nil(t, err)
-	time.Sleep(20 * time.Millisecond)
-	assert.DeepEqual(t, atomic.LoadInt32(&closeFlag), int32(0))
-	assert.Nil(t, con.Close())
-	time.Sleep(20 * time.Millisecond)
-	assert.DeepEqual(t, atomic.LoadInt32(&closeFlag), int32(1))
-}
-
-func TestWithSenseClientDisconnectionAndWithOnConnect(t *testing.T) {
-	var closeFlag int32
-	h := New(WithHostPorts("127.0.0.1:0"), WithSenseClientDisconnection(true), WithOnConnect(func(ctx context.Context, conn network.Conn) context.Context {
-		return ctx
-	}))
-	h.GET("/ping", func(c context.Context, ctx *app.RequestContext) {
-		assert.DeepEqual(t, "aa", string(ctx.Host()))
-		ch := make(chan struct{})
-		select {
-		case <-c.Done():
-			atomic.StoreInt32(&closeFlag, 1)
-			assert.DeepEqual(t, context.Canceled, c.Err())
-		case <-ch:
-		}
-	})
-	go h.Spin()
-	waitEngineRunning(h)
-
-	con, err := net.Dial("tcp", testutils.GetListenerAddr(h))
-	assert.Nil(t, err)
-	_, err = con.Write([]byte("GET /ping HTTP/1.1\r\nHost: aa\r\n\r\n"))
-	assert.Nil(t, err)
-	time.Sleep(20 * time.Millisecond)
-	assert.DeepEqual(t, atomic.LoadInt32(&closeFlag), int32(0))
-	assert.Nil(t, con.Close())
-	time.Sleep(20 * time.Millisecond)
-	assert.DeepEqual(t, atomic.LoadInt32(&closeFlag), int32(1))
-}
diff --git a/pkg/network/standard/connection_test.go b/pkg/network/standard/connection_test.go
index fa950f3..f0796cf 100644
--- a/pkg/network/standard/connection_test.go
+++ b/pkg/network/standard/connection_test.go
@@ -18,6 +18,7 @@ package standard
 
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"errors"
 	"io"
@@ -194,6 +195,34 @@ func TestReadBytes(t *testing.T) {
 	}
 }
 
+func TestStatefulConnConnectionCloseDetection(t *testing.T) {
+	cliConn, svrConn := net.Pipe()
+	svrCtx, cancelCtx := context.WithCancel(context.Background())
+	svrStatefulConn := NewStatefulConn(newConn(svrConn, 4096), newOnReadErrorFunc(cancelCtx)).(*statefulConn)
+	defer svrStatefulConn.Close()
+
+	// 1. normal request
+	svrStatefulConn.DetectConnectionClose()
+	svrStatefulConn.AbortBlockingRead()
+	select {
+	case <-svrCtx.Done():
+		t.Fatal("ctx should not be canceled")
+	default:
+	}
+
+	// 2. client close conn
+	svrStatefulConn.DetectConnectionClose()
+	cliConn.Close()
+	time.Sleep(100 * time.Millisecond) // wait a while
+
+	select {
+	case <-svrCtx.Done():
+		// expected
+	default:
+		t.Fatal("ctx should be canceled")
+	}
+}
+
 func TestWriteLogic(t *testing.T) {
 	c := mockConn{}
 	conn := newConn(&c, 4096)
diff --git a/pkg/network/standard/dial_test.go b/pkg/network/standard/dial_test.go
index 0c7e854..66f0fcf 100644
--- a/pkg/network/standard/dial_test.go
+++ b/pkg/network/standard/dial_test.go
@@ -40,7 +40,7 @@ func getListenerAddr(trans network.Transporter) string {
 
 func TestDial(t *testing.T) {
 	const nw = "tcp"
-	var addr = "127.0.0.1:0"
+	addr := "127.0.0.1:0"
 	transporter := NewTransporter(&config.Options{
 		Addr:    addr,
 		Network: nw,
@@ -65,7 +65,7 @@ func TestDial(t *testing.T) {
 
 func TestDialTLS(t *testing.T) {
 	const nw = "tcp"
-	var addr = "127.0.0.1:0"
+	addr := "127.0.0.1:0"
 	data := []byte("abcdefg")
 	listened := make(chan net.Listener)
 	go func() {
diff --git a/pkg/network/standard/transport_test.go b/pkg/network/standard/transport_test.go
index 57698d4..a38861a 100644
--- a/pkg/network/standard/transport_test.go
+++ b/pkg/network/standard/transport_test.go
@@ -23,11 +23,13 @@ import (
 	"testing"
 	"time"
 
+	internalNetwork "github.com/cloudwego/hertz/internal/network"
+
 	"github.com/cloudwego/hertz/pkg/common/config"
 	"github.com/cloudwego/hertz/pkg/network"
 )
 
-func assertWriteRead(t *testing.T, c io.ReadWriter, w string, r string) {
+func assertWriteRead(t *testing.T, c io.ReadWriter, w, r string) {
 	t.Helper()
 	_, err := c.Write([]byte(w))
 	if err != nil {
@@ -47,8 +49,12 @@ func TestTransporter(t *testing.T) {
 	handlerExit := make(chan struct{})
 	req := "hello"
 	resp := "world"
-	trans := NewTransporter(&config.Options{Network: "tcp", Addr: "127.0.0.1:0"}).(*transport)
+	trans := NewTransporter(&config.Options{Network: "tcp", Addr: "127.0.0.1:0", SenseClientDisconnection: true}).(*transport)
 	go trans.ListenAndServe(func(ctx context.Context, conn interface{}) error {
+		_, isStatefulConn := conn.(internalNetwork.StatefulConn)
+		if !isStatefulConn {
+			t.Fatal("SenseClientDisconnection configure failed")
+		}
 		c := conn.(network.Conn)
 		defer c.Close()
 		assertWriteRead(t, c, resp, req)
diff --git a/pkg/protocol/http1/server_test.go b/pkg/protocol/http1/server_test.go
index d478b36..a273a51 100644
--- a/pkg/protocol/http1/server_test.go
+++ b/pkg/protocol/http1/server_test.go
@@ -25,6 +25,8 @@ import (
 	"testing"
 	"time"
 
+	internalNetwork "github.com/cloudwego/hertz/internal/network"
+
 	inStats "github.com/cloudwego/hertz/internal/stats"
 	"github.com/cloudwego/hertz/pkg/app"
 	errs "github.com/cloudwego/hertz/pkg/common/errors"
@@ -400,6 +402,139 @@ func TestExpect100ContinueHandler(t *testing.T) {
 	assert.DeepEqual(t, "", string(response.Body()))
 }
 
+func TestSenseClientConnClose(t *testing.T) {
+	type connstate struct {
+		detectCalled bool
+		abortCalled  bool
+	}
+	reset := func(cs *connstate) {
+		cs.detectCalled = false
+		cs.abortCalled = false
+	}
+	state := &connstate{}
+
+	var (
+		detectFunc = func() {
+			state.detectCalled = true
+		}
+		abortFunc = func() {
+			state.abortCalled = true
+		}
+	)
+
+	server := &Server{}
+	reqCtx := &app.RequestContext{}
+	server.Core = &mockCore{
+		ctxPool: &sync.Pool{New: func() interface{} {
+			return reqCtx
+		}},
+		isRunning:   true,
+		mockHandler: func(c context.Context, ctx *app.RequestContext) {},
+	}
+
+	// normal
+	conn := mock.NewConn("GET / HTTP/1.1\nHost: foobar.com\n\n")
+	statefulConn := &mockStatefulConn{
+		conn,
+		nil,
+		detectFunc,
+		abortFunc,
+	}
+	server.Serve(context.Background(), statefulConn)
+	assert.True(t, state.detectCalled)
+	assert.True(t, state.abortCalled)
+	reset(state)
+
+	// 100 continue
+	conn = mock.NewConn("POST /foo HTTP/1.1\r\nHost: gle.com\r\nExpect: 100-continue\r\nContent-Length: 5\r\nContent-Type: a/b\r\n\r\n12345")
+	statefulConn.Conn = conn
+	server.Serve(context.Background(), statefulConn)
+	assert.True(t, state.detectCalled)
+	assert.True(t, state.abortCalled)
+	reset(state)
+
+	// 100 continue: error
+	conn = mock.NewConn("POST /foo HTTP/1.1\r\n" +
+		"Host: gle.com\r\n" +
+		"Expect: 100-continue\r\n" +
+		"Content-Length: 40\r\n" +
+		"Content-Type: multipart/form-data; boundary=1\r\n" +
+		"\r\n" +
+		"--1\r\n" +
+		"broken-multipart-body",
+	)
+	statefulConn.Conn = conn
+	server.Serve(context.Background(), statefulConn)
+	assert.False(t, state.detectCalled)
+	assert.False(t, state.abortCalled)
+	reset(state)
+
+	// bodyStream
+	server.StreamRequestBody = true
+	conn = mock.NewConn("POST /foo HTTP/1.1\r\nHost: gle.com\r\nExpect: 100-continue\r\nContent-Length: 0\r\nContent-Type: a/b\r\n\r\n12345")
+	statefulConn.Conn = conn
+	server.Serve(context.Background(), statefulConn)
+	assert.False(t, state.detectCalled)
+	assert.False(t, state.abortCalled)
+	reset(state)
+	server.StreamRequestBody = false
+
+	// hijacked
+	conn = mock.NewConn("GET / HTTP/1.1\nHost: foobar.com\n\n")
+	statefulConn = &mockStatefulConn{
+		conn,
+		nil,
+		detectFunc,
+		abortFunc,
+	}
+	server.HijackConnHandle = func(c network.Conn, h app.HijackHandler) {
+		h(c)
+	}
+	server.Core = &mockCore{
+		ctxPool: &sync.Pool{New: func() interface{} {
+			return reqCtx
+		}},
+		isRunning: true,
+		mockHandler: func(c context.Context, ctx *app.RequestContext) {
+			ctx.SetHijackHandler(func(conn network.Conn) {
+				_ = conn.Len()
+			})
+		},
+	}
+	server.Serve(context.Background(), statefulConn)
+	assert.True(t, state.detectCalled)
+	assert.True(t, state.abortCalled)
+	reset(state)
+	server.HijackConnHandle = nil
+
+	// handler panic
+	conn = mock.NewConn("GET / HTTP/1.1\nHost: foobar.com\n\n")
+	statefulConn = &mockStatefulConn{
+		conn,
+		nil,
+		detectFunc,
+		abortFunc,
+	}
+	server.Core = &mockCore{
+		ctxPool: &sync.Pool{New: func() interface{} {
+			return reqCtx
+		}},
+		isRunning: true,
+		mockHandler: func(c context.Context, ctx *app.RequestContext) {
+			// panic should be recovered by the recovery middleware, just mock one here.
+			// perform like a normal request
+			defer func() {
+				recover()
+			}()
+			panic("mock panic")
+		},
+	}
+	server.Serve(context.Background(), statefulConn)
+	assert.True(t, state.detectCalled)
+	assert.True(t, state.abortCalled)
+	reset(state)
+}
+
 type mockController struct {
 	FinishTimes int
 }
@@ -494,3 +629,24 @@ func TestShouldRecordInTraceError(t *testing.T) {
 	assert.True(t, shouldRecordInTraceError(errTimeout))
 	assert.True(t, shouldRecordInTraceError(errors.New("foo error")))
 }
+
+var _ internalNetwork.StatefulConn = &mockStatefulConn{}
+
+type mockStatefulConn struct {
+	network.Conn
+	Ctx                       context.Context
+	DetectConnectionCloseFunc func()
+	AbortBlockingReadFunc     func()
+}
+
+func (c *mockStatefulConn) DetectConnectionClose() {
+	c.DetectConnectionCloseFunc()
+}
+
+func (c *mockStatefulConn) AbortBlockingRead() {
+	c.AbortBlockingReadFunc()
+}
+
+func (c *mockStatefulConn) Context() context.Context {
+	return c.Ctx
+}