feat(client): add optional pooled connection health check

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,62 @@
+// 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() bool
+}
+
+// IsHealthy checks whether an idle pooled connection is safe to reuse without
+// consuming application data.
+func (c *Conn) IsHealthy(timeout time.Duration) bool {
+	if c == nil || c.Conn == nil || timeout <= 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()
+	}
+
+	// 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(timeout); 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()
+}

pkg/network/netpoll/healthcheck_linux_test.go

@@ -0,0 +1,71 @@
+// 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 (
+	"testing"
+	"time"
+
+	"github.com/cloudwego/hertz/pkg/common/test/mock"
+	"github.com/cloudwego/hertz/pkg/network"
+)
+
+type reuseHealthProbe struct {
+	network.Conn
+	healthy bool
+	calls   int
+}
+
+func (p *reuseHealthProbe) IsHealthyForReuse() bool {
+	p.calls++
+	return p.healthy
+}
+
+func TestConnIsHealthyDelegatesToOwnerProbe(t *testing.T) {
+	for _, tt := range []struct {
+		name    string
+		healthy bool
+	}{
+		{name: "healthy", healthy: true},
+		{name: "unhealthy", healthy: false},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			probe := &reuseHealthProbe{Conn: mock.NewConn(""), healthy: tt.healthy}
+			conn := &Conn{Conn: probe}
+			if got := conn.IsHealthy(50 * time.Microsecond); got != tt.healthy {
+				t.Fatalf("owner probe result: got %v, want %v", got, tt.healthy)
+			}
+			if probe.calls != 1 {
+				t.Fatalf("owner probe calls: got %d, want 1", probe.calls)
+			}
+		})
+	}
+}
+
+func TestConnIsHealthyFallsBackToTimedPeek(t *testing.T) {
+	conn := &Conn{Conn: mock.NewConn("")}
+	if !conn.IsHealthy(50 * time.Microsecond) {
+		t.Fatal("old netpoll without owner probe should use the timed Peek fallback")
+	}
+}
+
+func TestConnIsHealthyFallbackRejectsStaleConnection(t *testing.T) {
+	conn := &Conn{Conn: mock.NewBrokenConn("")}
+	if conn.IsHealthy(50 * time.Microsecond) {
+		t.Fatal("timed Peek fallback must reject a stale connection")
+	}
+}