feat(client): add optional pooled connection health check

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/bytedance/gopkg v0.1.4
 	github.com/bytedance/sonic v1.15.0
 	github.com/cloudwego/gopkg v0.2.0
-	github.com/cloudwego/netpoll v0.7.3
+	github.com/cloudwego/netpoll v0.7.5-0.20260730165643-598bbe1df4d1
 	github.com/fsnotify/fsnotify v1.5.4
 	github.com/stretchr/testify v1.10.0
 	github.com/tidwall/gjson v1.14.4
@@ -28,3 +28,5 @@ require (
 	golang.org/x/arch v0.0.0-20210923205945-b76863e36670 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
+
+replace github.com/cloudwego/netpoll => github.com/StellarisW/netpoll v0.7.5-0.20260730165643-598bbe1df4d1

go.sum

@@ -1,3 +1,5 @@
+github.com/StellarisW/netpoll v0.7.5-0.20260730165643-598bbe1df4d1 h1:AWqDpjvT9ivpyM/qnk8lu+7ZRN0n2ru+9fGfAZvHHa4=
+github.com/StellarisW/netpoll v0.7.5-0.20260730165643-598bbe1df4d1/go.mod h1:KiNpLI5MX9vR0xj4gKqyioOrHlp8G0XBMqIV9HsvMCc=
 github.com/bytedance/gopkg v0.1.4 h1:oZnQwnX82KAIWb7033bEwtxvTqXcYMxDBaQxo5JJHWM=
 github.com/bytedance/gopkg v0.1.4/go.mod h1:v1zWfPm21Fb+OsyXN2VAHdL6TBb2L88anLQgdyje6R4=
 github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE=
@@ -8,8 +10,6 @@ github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI
 github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU=
 github.com/cloudwego/gopkg v0.2.0 h1:EU8Ahrj0rCfKZQdah50zKnlrQ1o2AdPYM87UclIqLME=
 github.com/cloudwego/gopkg v0.2.0/go.mod h1:WjQPYI8PesfQalIVcLzVJBb1EAopioZ+D+3UGJ+dNBs=
-github.com/cloudwego/netpoll v0.7.3 h1:E9ImEseXM9BdHS+5aLxcE9Z0c7okFbM11XMwwJ00LxY=
-github.com/cloudwego/netpoll v0.7.3/go.mod h1:KiNpLI5MX9vR0xj4gKqyioOrHlp8G0XBMqIV9HsvMCc=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=

pkg/app/client/client.go

@@ -669,6 +669,7 @@ func newHttp1OptionFromClient(c *Client) *http1.ClientOptions {
 		MaxConns:                      c.options.MaxConnsPerHost,
 		MaxConnDuration:               c.options.MaxConnDuration,
 		MaxIdleConnDuration:           c.options.MaxIdleConnDuration,
+		PooledConnHealthCheck:         c.options.PooledConnHealthCheck,
 		ReadTimeout:                   c.options.ReadTimeout,
 		WriteTimeout:                  c.options.WriteTimeout,
 		MaxResponseBodySize:           c.options.MaxResponseBodySize,

pkg/app/client/client_test.go

@@ -201,6 +201,127 @@ func TestCloseIdleTLSConnections(t *testing.T) {
 	}
 }
 
+func TestPooledConnHealthCheckTLS(t *testing.T) {
+	var accepted, requests int32
+	firstIdleClosed := make(chan struct{})
+	var closeFirstIdle sync.Once
+
+	httpsServer := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != consts.MethodPost {
+			http.Error(w, "unexpected method", http.StatusMethodNotAllowed)
+			return
+		}
+		body, err := io.ReadAll(r.Body)
+		if err != nil || string(body) != "payload" {
+			http.Error(w, "unexpected body", http.StatusBadRequest)
+			return
+		}
+		atomic.AddInt32(&requests, 1)
+		_, _ = w.Write([]byte("ok"))
+	}))
+	httpsServer.Config.ConnState = func(conn net.Conn, state http.ConnState) {
+		switch state {
+		case http.StateNew:
+			atomic.AddInt32(&accepted, 1)
+		case http.StateIdle:
+			if atomic.LoadInt32(&requests) == 1 {
+				closeFirstIdle.Do(func() {
+					_ = conn.Close()
+					close(firstIdleClosed)
+				})
+			}
+		}
+	}
+	httpsServer.StartTLS()
+	defer httpsServer.Close()
+
+	c, err := NewClient(
+		WithTLSConfig(httpsServer.Client().Transport.(*http.Transport).TLSClientConfig),
+		WithPooledConnHealthCheck(true),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.CloseIdleConnections()
+
+	doPost := func() {
+		req := protocol.AcquireRequest()
+		defer protocol.ReleaseRequest(req)
+		req.Header.SetMethod(consts.MethodPost)
+		req.SetRequestURI(httpsServer.URL + "/health-check")
+		req.SetBodyString("payload")
+		resp := protocol.AcquireResponse()
+		defer protocol.ReleaseResponse(resp)
+
+		if err := c.Do(context.Background(), req, resp); err != nil {
+			t.Fatal(err)
+		}
+		assert.DeepEqual(t, http.StatusOK, resp.StatusCode())
+		assert.DeepEqual(t, "ok", string(resp.Body()))
+	}
+
+	doPost()
+	select {
+	case <-firstIdleClosed:
+	case <-time.After(time.Second):
+		t.Fatal("server did not close the first idle TLS connection")
+	}
+	doPost()
+
+	assert.DeepEqual(t, int32(2), atomic.LoadInt32(&accepted))
+	assert.DeepEqual(t, int32(2), atomic.LoadInt32(&requests))
+}
+
+func TestPooledConnHealthCheckReusesTLS(t *testing.T) {
+	var accepted, requests int32
+	httpsServer := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != consts.MethodPost {
+			http.Error(w, "unexpected method", http.StatusMethodNotAllowed)
+			return
+		}
+		body, err := io.ReadAll(r.Body)
+		if err != nil || string(body) != "payload" {
+			http.Error(w, "unexpected body", http.StatusBadRequest)
+			return
+		}
+		atomic.AddInt32(&requests, 1)
+		_, _ = w.Write([]byte("ok"))
+	}))
+	httpsServer.Config.ConnState = func(_ net.Conn, state http.ConnState) {
+		if state == http.StateNew {
+			atomic.AddInt32(&accepted, 1)
+		}
+	}
+	httpsServer.StartTLS()
+	defer httpsServer.Close()
+
+	c, err := NewClient(
+		WithTLSConfig(httpsServer.Client().Transport.(*http.Transport).TLSClientConfig),
+		WithPooledConnHealthCheck(true),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.CloseIdleConnections()
+
+	for i := 0; i < 2; i++ {
+		req := protocol.AcquireRequest()
+		req.Header.SetMethod(consts.MethodPost)
+		req.SetRequestURI(httpsServer.URL + "/health-check")
+		req.SetBodyString("payload")
+		resp := protocol.AcquireResponse()
+		err := c.Do(context.Background(), req, resp)
+		assert.Nil(t, err)
+		assert.DeepEqual(t, http.StatusOK, resp.StatusCode())
+		assert.DeepEqual(t, "ok", string(resp.Body()))
+		protocol.ReleaseRequest(req)
+		protocol.ReleaseResponse(resp)
+	}
+
+	assert.DeepEqual(t, int32(1), atomic.LoadInt32(&accepted))
+	assert.DeepEqual(t, int32(2), atomic.LoadInt32(&requests))
+}
+
 func TestClientInvalidURI(t *testing.T) {
 	opt, ln := newTestOptions(t)
 	defer ln.Close()
@@ -2290,6 +2411,14 @@ func TestClientHostClientConfigHook(t *testing.T) {
 	assert.DeepEqual(t, "FOO.BAR:443", hcr.Addr)
 }
 
+func TestPooledConnHealthCheckOptionPropagation(t *testing.T) {
+	c, err := NewClient(WithPooledConnHealthCheck(true))
+	assert.Nil(t, err)
+
+	http1Options := newHttp1OptionFromClient(c)
+	assert.DeepEqual(t, true, http1Options.PooledConnHealthCheck)
+}
+
 func TestClientDialerName(t *testing.T) {
 	client, _ := NewClient()
 	dName, err := client.GetDialerName()

pkg/app/client/option.go

@@ -49,6 +49,17 @@ func WithMaxIdleConnDuration(t time.Duration) config.ClientOption {
 	}}
 }
 
+// WithPooledConnHealthCheck determines whether idle pooled connections are
+// probed before reuse. Unhealthy connections are discarded before writing a
+// request.
+//
+// The check uses a short read probe and is disabled by default.
+func WithPooledConnHealthCheck(enabled bool) config.ClientOption {
+	return config.ClientOption{F: func(o *config.ClientOptions) {
+		o.PooledConnHealthCheck = enabled
+	}}
+}
+
 // WithMaxConnDuration sets max connection duration, keep-alive connections are closed after this duration.
 func WithMaxConnDuration(t time.Duration) config.ClientOption {
 	return config.ClientOption{F: func(o *config.ClientOptions) {

pkg/app/client/option_test.go

@@ -33,6 +33,7 @@ func TestClientOptions(t *testing.T) {
 	assert.DeepEqual(t, 0, opt.MaxConnsPerHost)
 	assert.DeepEqual(t, consts.DefaultDialTimeout, opt.DialTimeout)
 	assert.DeepEqual(t, consts.DefaultMaxIdleConnDuration, opt.MaxIdleConnDuration)
+	assert.DeepEqual(t, false, opt.PooledConnHealthCheck)
 	assert.DeepEqual(t, true, opt.KeepAlive)
 	assert.DeepEqual(t, 5*time.Second, opt.ObservationInterval)
 
@@ -41,6 +42,7 @@ func TestClientOptions(t *testing.T) {
 		WithDialTimeout(100 * time.Millisecond),
 		WithMaxConnsPerHost(128),
 		WithMaxIdleConnDuration(5 * time.Second),
+		WithPooledConnHealthCheck(true),
 		WithMaxConnDuration(10 * time.Second),
 		WithMaxConnWaitTimeout(5 * time.Second),
 		WithKeepAlive(false),
@@ -59,6 +61,7 @@ func TestClientOptions(t *testing.T) {
 	assert.DeepEqual(t, 100*time.Millisecond, opt.DialTimeout)
 	assert.DeepEqual(t, 128, opt.MaxConnsPerHost)
 	assert.DeepEqual(t, 5*time.Second, opt.MaxIdleConnDuration)
+	assert.DeepEqual(t, true, opt.PooledConnHealthCheck)
 	assert.DeepEqual(t, 10*time.Second, opt.MaxConnDuration)
 	assert.DeepEqual(t, 5*time.Second, opt.MaxConnWaitTimeout)
 	assert.DeepEqual(t, false, opt.KeepAlive)

pkg/common/config/client_option.go

@@ -56,12 +56,15 @@ type ClientOptions struct {
 	MaxConnsPerHost int
 
 	MaxIdleConnDuration time.Duration
-	MaxConnDuration     time.Duration
-	MaxConnWaitTimeout  time.Duration
-	KeepAlive           bool
-	ReadTimeout         time.Duration
-	TLSConfig           *tls.Config
-	ResponseBodyStream  bool
+	// PooledConnHealthCheck determines whether an idle pooled connection is
+	// probed before reuse. It is disabled by default.
+	PooledConnHealthCheck bool
+	MaxConnDuration       time.Duration
+	MaxConnWaitTimeout    time.Duration
+	KeepAlive             bool
+	ReadTimeout           time.Duration
+	TLSConfig             *tls.Config
+	ResponseBodyStream    bool
 
 	// Client name. Used in User-Agent request header.
 	//

pkg/network/netpoll/healthcheck_linux.go

@@ -0,0 +1,64 @@
+// Copyright 2022 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.
+
+//go:build linux
+
+package netpoll
+
+import (
+	"errors"
+	"net"
+	"time"
+
+	errs "github.com/cloudwego/hertz/pkg/common/errors"
+)
+
+type reusableConnHealthChecker interface {
+	IsHealthyForReuse(ownerWaitTimeout time.Duration) bool
+}
+
+// IsHealthy checks whether an idle pooled connection is safe to reuse without
+// consuming application data. probeTimeout bounds the fallback timed Peek;
+// ownerWaitTimeout bounds waiting for the netpoll operator. The owner-side
+// probe itself is non-blocking once acquired.
+func (c *Conn) IsHealthy(probeTimeout, ownerWaitTimeout time.Duration) bool {
+	if c == nil || c.Conn == nil || probeTimeout <= 0 || ownerWaitTimeout <= 0 || c.Len() > 0 {
+		return false
+	}
+
+	// Newer netpoll versions synchronize this check with their receive owner.
+	// Keep the narrow assertion here so Hertz remains compatible with older
+	// netpoll versions and third-party network.Conn implementations.
+	if checker, ok := c.Conn.(reusableConnHealthChecker); ok {
+		return checker.IsHealthyForReuse(ownerWaitTimeout)
+	}
+
+	// Older netpoll versions have no owner-side probe. Preserve the bounded
+	// Peek fallback for rolling upgrades instead of inspecting their raw fd.
+	if err := c.SetReadTimeout(probeTimeout); err != nil {
+		return false
+	}
+	p, err := c.Peek(1)
+	if resetErr := c.SetReadTimeout(0); resetErr != nil {
+		return false
+	}
+	if len(p) != 0 {
+		return false
+	}
+	if errors.Is(err, errs.ErrTimeout) {
+		return true
+	}
+	var timeoutErr net.Error
+	return errors.As(err, &timeoutErr) && timeoutErr.Timeout()
+}