Repository: cloudwego/hertz
Pull request #871 (2023-07-24 09:32:10)

optimize(http1): retry if the error is caused by broken pooled conn
#### What type of PR is this?
optimize

#### Check the PR title.
- [x] This PR title match the format: `<type>(optional scope): <description>`.
- [x] 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.
HTTP/1.1 client 增加对于连接池中连接被提前关闭导致的错误请求的重试


#### (Optional) More detail description for this PR(en: English/zh: Chinese).
<!--
Provide more detailed info for review. If it is a perf type PR, perf data is suggested to give.
-->
en: 
Since hertz does not retry by default, once the connection in the connection pool is closed by the peer in advance, the client's request will directly fail this time, which is unreasonable.
This optimization adds a default retry logic for such scenarios to ensure that no more requests will fail due to problems with the connection pool itself.

zh(optional):
由于hertz默认是不重试的，一旦连接池中的连接被对方提前关闭，那么客户端这次的请求将直接失败，这是不合理的。
此次优化针对此类场景添加了默认的重试逻辑，保证不会再有请求因为连接池本身的问题而失败。
#### (Optional) Which issue(s) this PR fixes:


#### (Optional) The PR that updates user documentation:

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/cloudwego/hertz/pull/871?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) Report
Patch coverage: **`58.33%`** and project coverage change: **`-0.05%`** :warning:
> Comparison is base [(`3161f04`)](https://app.codecov.io/gh/cloudwego/hertz/commit/3161f043a97d062f887fa2d8a9af9ca9bc96fc95?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) 79.74% compared to head [(`8e839df`)](https://app.codecov.io/gh/cloudwego/hertz/pull/871?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) 79.69%.

> :exclamation: Current head 8e839df differs from pull request most recent head 00aaf86. Consider uploading reports for the commit 00aaf86 to get more accurate results

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


```diff
@@             Coverage Diff             @@
##           develop     #871      +/-   ##
===========================================
- Coverage    79.74%   79.69%   -0.05%     
===========================================
  Files           97       97              
  Lines         9534     9570      +36     
===========================================
+ Hits          7603     7627      +24     
- Misses        1432     1442      +10     
- Partials       499      501       +2     
```


| [Files Changed](https://app.codecov.io/gh/cloudwego/hertz/pull/871?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) | Coverage Δ | |
|---|---|---|
| [pkg/common/errors/errors.go](https://app.codecov.io/gh/cloudwego/hertz/pull/871?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2NvbW1vbi9lcnJvcnMvZXJyb3JzLmdv) | `87.95% <ø> (ø)` | |
| [pkg/protocol/http1/server.go](https://app.codecov.io/gh/cloudwego/hertz/pull/871?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL2h0dHAxL3NlcnZlci5nbw==) | `74.27% <ø> (ø)` | |
| [pkg/common/test/mock/network.go](https://app.codecov.io/gh/cloudwego/hertz/pull/871?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2NvbW1vbi90ZXN0L21vY2svbmV0d29yay5nbw==) | `78.00% <12.50%> (-8.03%)` | :arrow_down: |
| [pkg/protocol/http1/client.go](https://app.codecov.io/gh/cloudwego/hertz/pull/871?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL2h0dHAxL2NsaWVudC5nbw==) | `73.28% <81.25%> (+0.16%)` | :arrow_up: |

... and [2 files with indirect coverage changes](https://app.codecov.io/gh/cloudwego/hertz/pull/871/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego)


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/cloudwego/hertz/pull/871?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).


Patch:
diff --git a/pkg/app/context.go b/pkg/app/context.go
index e4fcd0a..559b84d 100644
--- a/pkg/app/context.go
+++ b/pkg/app/context.go
@@ -754,7 +754,8 @@ func (ctx *RequestContext) ResetWithoutConn() {
 		close(ctx.finished)
 		ctx.finished = nil
 	}
-	ctx.Request.Reset()
+
+	ctx.Request.ResetWithoutConn()
 	ctx.Response.Reset()
 	if ctx.IsEnableTrace() {
 		ctx.traceInfo.Reset()
diff --git a/pkg/common/errors/errors.go b/pkg/common/errors/errors.go
index fdaf2d5..918c7fa 100644
--- a/pkg/common/errors/errors.go
+++ b/pkg/common/errors/errors.go
@@ -53,17 +53,15 @@ var (
 	ErrChunkedStream      = errors.New("chunked stream")
 	ErrBodyTooLarge       = errors.New("body size exceeds the given limit")
 	ErrHijacked           = errors.New("connection has been hijacked")
-	ErrIdleTimeout        = errors.New("idle timeout")
 	ErrTimeout            = errors.New("timeout")
-	ErrReadTimeout        = errors.New("read timeout")
-	ErrWriteTimeout       = errors.New("write timeout")
-	ErrDialTimeout        = errors.New("dial timeout")
+	ErrIdleTimeout        = errors.New("idle timeout")
 	ErrNothingRead        = errors.New("nothing read")
 	ErrShortConnection    = errors.New("short connection")
 	ErrNoFreeConns        = errors.New("no free connections available to host")
 	ErrConnectionClosed   = errors.New("connection closed")
 	ErrNotSupportProtocol = errors.New("not support protocol")
 	ErrNoMultipartForm    = errors.New("request has no multipart/form-data Content-Type")
+	ErrBadPoolConn        = errors.New("connection is closed by peer while being in the connection pool")
 )
 
 // ErrorType is an unsigned 64-bit error code as defined in the hertz spec.
diff --git a/pkg/protocol/client/client.go b/pkg/protocol/client/client.go
index 3715255..777f55c 100644
--- a/pkg/protocol/client/client.go
+++ b/pkg/protocol/client/client.go
@@ -43,7 +43,6 @@ package client
 
 import (
 	"context"
-	"io"
 	"sync"
 	"time"
 
@@ -88,16 +87,6 @@ func DefaultRetryIf(req *protocol.Request, resp *protocol.Response, err error) b
 	if isIdempotent(req, resp, err) {
 		return true
 	}
-	// Retry non-idempotent requests if the server closes
-	// the connection before sending the response.
-	//
-	// This case is possible if the server closes the idle
-	// keep-alive connection on timeout.
-	//
-	// Apache and nginx usually do this.
-	if err == io.EOF {
-		return true
-	}
 
 	return false
 }
diff --git a/pkg/protocol/http1/client.go b/pkg/protocol/http1/client.go
index 34bb5f0..c306f07 100644
--- a/pkg/protocol/http1/client.go
+++ b/pkg/protocol/http1/client.go
@@ -51,6 +51,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"syscall"
 	"time"
 
 	"github.com/cloudwego/hertz/internal/bytesconv"
@@ -362,6 +363,7 @@ func (c *HostClient) Do(ctx context.Context, req *protocol.Request, resp *protoc
 		canIdempotentRetry bool
 		isDefaultRetryFunc                    = true
 		attempts           uint               = 0
+		connAttempts       uint               = 0
 		maxAttempts        uint               = 1
 		isRequestRetryable client.RetryIfFunc = client.DefaultRetryIf
 	)
@@ -380,17 +382,38 @@ func (c *HostClient) Do(ctx context.Context, req *protocol.Request, resp *protoc
 	atomic.AddInt32(&c.pendingRequests, 1)
 	req.Options().StartRequest()
 	for {
+		select {
+		case <-ctx.Done():
+			req.CloseBodyStream() //nolint:errcheck
+			return ctx.Err()
+		default:
+		}
+
 		canIdempotentRetry, err = c.do(req, resp)
 		// If there is no custom retry and err is equal to nil, the loop simply exits.
 		if err == nil && isDefaultRetryFunc {
+			if connAttempts != 0 {
+				hlog.SystemLogger().Warnf("Client connection attempt times: %d, url: %s. "+
+					"This is mainly because the connection in pool is closed by peer in advance. "+
+					"If this number is too high which indicates that long-connection are basically unavailable, "+
+					"try to change the request to short-connection.\n", connAttempts, req.URI().FullURI())
+			}
 			break
 		}
 
+		// This connection is closed by the peer when it is in the connection pool.
+		//
+		// This case is possible if the server closes the idle
+		// keep-alive connection on timeout.
+		//
+		// Apache and nginx usually do this.
+		if canIdempotentRetry && client.DefaultRetryIf(req, resp, err) && errors.Is(err, errs.ErrBadPoolConn) {
+			connAttempts++
+			continue
+		}
+
 		if isDefaultRetryFunc {
-			// canIdempotentRetry only makes sense if the user hasn't provided a custom retry function.
-			if !canIdempotentRetry {
-				break
-			}
+			break
 		}
 
 		attempts++
@@ -516,7 +539,7 @@ func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Respo
 	if reqTimeout < dialTimeout || dialTimeout == 0 {
 		dialTimeout = reqTimeout
 	}
-	cc, err := c.acquireConn(dialTimeout)
+	cc, inPool, err := c.acquireConn(dialTimeout)
 	// if getting connection error, fast fail
 	if err != nil {
 		return false, err
@@ -588,13 +611,16 @@ func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Respo
 		// Only if the connection is closed while writing the request. Try to parse the response and return.
 		// In this case, the request/response is considered as successful.
 		// Otherwise, return the former error.
-
 		zr := c.acquireReader(conn)
 		defer zr.Release()
 		if respI.ReadHeaderAndLimitBody(resp, zr, c.MaxResponseBodySize) == nil {
 			return false, nil
 		}
 
+		if inPool {
+			err = errs.ErrBadPoolConn
+		}
+
 		return true, err
 	}
 
@@ -620,6 +646,24 @@ func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Respo
 	}
 	zr := c.acquireReader(conn)
 
+	// errs.ErrBadPoolConn error are returned when the
+	// 1 byte peek read fails, and we're actually anticipating a response.
+	// Usually this is just due to the inherent keep-alive shut down race,
+	// where the server closed the connection at the same time the client
+	// wrote. The underlying err field is usually io.EOF or some
+	// ECONNRESET sort of thing which varies by platform.
+	_, err = zr.Peek(1)
+	if err != nil {
+		zr.Release() //nolint:errcheck
+		c.closeConn(cc)
+		if inPool && (err == io.EOF || err == syscall.ECONNRESET) {
+			return true, errs.ErrBadPoolConn
+		}
+		// if this is not a pooled connection,
+		// we should not retry to avoid getting stuck in an endless retry loop.
+		return false, err
+	}
+
 	// init here for passing in ReadBodyStream's closure
 	// and this value will be assigned after reading Response's Header
 	//
@@ -676,7 +720,7 @@ func (c *HostClient) SetMaxConns(newMaxConns int) {
 	c.connsLock.Unlock()
 }
 
-func (c *HostClient) acquireConn(dialTimeout time.Duration) (cc *clientConn, err error) {
+func (c *HostClient) acquireConn(dialTimeout time.Duration) (cc *clientConn, inPool bool, err error) {
 	createConn := false
 	startCleaner := false
 
@@ -705,11 +749,11 @@ func (c *HostClient) acquireConn(dialTimeout time.Duration) (cc *clientConn, err
 	c.connsLock.Unlock()
 
 	if cc != nil {
-		return cc, nil
+		return cc, true, nil
 	}
 	if !createConn {
 		if c.MaxConnWaitTimeout <= 0 {
-			return nil, errs.ErrNoFreeConns
+			return nil, true, errs.ErrNoFreeConns
 		}
 
 		timeout := c.MaxConnWaitTimeout
@@ -736,9 +780,9 @@ func (c *HostClient) acquireConn(dialTimeout time.Duration) (cc *clientConn, err
 
 		select {
 		case <-w.ready:
-			return w.conn, w.err
+			return w.conn, true, w.err
 		case <-tc.C:
-			return nil, errs.ErrNoFreeConns
+			return nil, true, errs.ErrNoFreeConns
 		}
 	}
 
@@ -749,11 +793,11 @@ func (c *HostClient) acquireConn(dialTimeout time.Duration) (cc *clientConn, err
 	conn, err := c.dialHostHard(dialTimeout)
 	if err != nil {
 		c.decConnsCount()
-		return nil, err
+		return nil, false, err
 	}
 	cc = acquireClientConn(conn)
 
-	return cc, nil
+	return cc, false, nil
 }
 
 func (c *HostClient) queueForIdle(w *wantConn) {
diff --git a/pkg/protocol/http1/server.go b/pkg/protocol/http1/server.go
index 8aaed6d..77cc74e 100644
--- a/pkg/protocol/http1/server.go
+++ b/pkg/protocol/http1/server.go
@@ -48,7 +48,7 @@ const NextProtoTLS = suite.HTTP1
 
 var (
 	errHijacked        = errs.New(errs.ErrHijacked, errs.ErrorTypePublic, nil)
-	errIdleTimeout     = errs.New(errs.ErrIdleTimeout, errs.ErrorTypePublic, nil)
+	errIdleTimeout     = errs.New(errs.ErrIdleTimeout, errs.ErrorTypePrivate, nil)
 	errShortConnection = errs.New(errs.ErrShortConnection, errs.ErrorTypePublic, "server is going to close the connection")
 	errUnexpectedEOF   = errs.NewPublic(io.ErrUnexpectedEOF.Error() + " when reading request")
 )
diff --git a/pkg/protocol/request.go b/pkg/protocol/request.go
index af29db1..ef70f92 100644
--- a/pkg/protocol/request.go
+++ b/pkg/protocol/request.go
@@ -176,16 +176,25 @@ func (req *Request) MayContinue() bool {
 	return bytes.Equal(req.Header.peek(bytestr.StrExpect), bytestr.Str100Continue)
 }
 
+// Scheme returns the scheme of the request.
+// uri will be parsed in ServeHTTP(before user's process), so that there is no need for uri nil-check.
 func (req *Request) Scheme() []byte {
 	return req.uri.Scheme()
 }
 
-func (req *Request) ResetSkipHeader() {
+// For keepalive connection reuse.
+// It is roughly the same as ResetSkipHeader, except that the connection related fields are removed:
+// - req.isTLS
+func (req *Request) resetSkipHeaderAndConn() {
 	req.ResetBody()
 	req.uri.Reset()
 	req.parsedURI = false
 	req.parsedPostArgs = false
 	req.postArgs.Reset()
+}
+
+func (req *Request) ResetSkipHeader() {
+	req.resetSkipHeaderAndConn()
 	req.isTLS = false
 }
 
@@ -814,6 +823,14 @@ func (req *Request) SetConnectionClose() {
 	req.Header.SetConnectionClose(true)
 }
 
+func (req *Request) ResetWithoutConn() {
+	req.Header.Reset()
+	req.resetSkipHeaderAndConn()
+	req.CloseBodyStream()
+
+	req.options = nil
+}
+
 // AcquireRequest returns an empty Request instance from request pool.
 //
 // The returned Request instance may be passed to ReleaseRequest when it is


Test patch:
diff --git a/pkg/app/client/client_test.go b/pkg/app/client/client_test.go
index ba00f78..8449cd4 100644
--- a/pkg/app/client/client_test.go
+++ b/pkg/app/client/client_test.go
@@ -453,7 +453,9 @@ func TestClientReadTimeout(t *testing.T) {
 		req.SetConnectionClose()
 
 		if err := c.Do(context.Background(), req, res); !errors.Is(err, errs.ErrTimeout) {
-			t.Errorf("expected ErrTimeout got %#v", err)
+			if !strings.Contains(err.Error(), "timeout") {
+				t.Errorf("expected ErrTimeout got %#v", err)
+			}
 		}
 
 		protocol.ReleaseRequest(req)
@@ -2267,7 +2269,7 @@ func TestClientDoWithDialFunc(t *testing.T) {
 
 func TestClientState(t *testing.T) {
 	opt := config.NewOptions([]config.Option{})
-	opt.Addr = ":11000"
+	opt.Addr = "127.0.0.1:11000"
 	engine := route.NewEngine(opt)
 	go engine.Run()
 
diff --git a/pkg/app/context_test.go b/pkg/app/context_test.go
index b9e9d2f..8bf52f4 100644
--- a/pkg/app/context_test.go
+++ b/pkg/app/context_test.go
@@ -654,8 +654,10 @@ func TestContextReset(t *testing.T) {
 	c.Params = param.Params{param.Param{}}
 	c.Error(errors.New("test")) // nolint: errcheck
 	c.Set("foo", "bar")
+	c.Request.SetIsTLS(true)
 	c.ResetWithoutConn()
-
+	c.Request.URI()
+	assert.DeepEqual(t, "https", string(c.Request.Scheme()))
 	assert.False(t, c.IsAborted())
 	assert.DeepEqual(t, 0, len(c.Errors))
 	assert.Nil(t, c.Errors.Errors())
diff --git a/pkg/app/server/hertz_test.go b/pkg/app/server/hertz_test.go
index b9a07fb..bc3e1a0 100644
--- a/pkg/app/server/hertz_test.go
+++ b/pkg/app/server/hertz_test.go
@@ -311,16 +311,17 @@ func TestNotAbsolutePathWithRawPath(t *testing.T) {
 }
 
 func TestWithBasePath(t *testing.T) {
-	engine := New(WithBasePath("/hertz"), WithHostPorts("127.0.0.1:9898"))
+	engine := New(WithBasePath("/hertz"), WithHostPorts("127.0.0.1:19898"))
 	engine.POST("/test", func(c context.Context, ctx *app.RequestContext) {
 	})
 	go engine.Run()
-	time.Sleep(200 * time.Microsecond)
+	time.Sleep(500 * time.Microsecond)
 	var r http.Request
 	r.ParseForm()
 	r.Form.Add("xxxxxx", "xxx")
 	body := strings.NewReader(r.Form.Encode())
-	resp, _ := http.Post("http://127.0.0.1:9898/hertz/test", "application/x-www-form-urlencoded", body)
+	resp, err := http.Post("http://127.0.0.1:19898/hertz/test", "application/x-www-form-urlencoded", body)
+	assert.Nil(t, err)
 	assert.DeepEqual(t, consts.StatusOK, resp.StatusCode)
 }
 
diff --git a/pkg/common/test/mock/network.go b/pkg/common/test/mock/network.go
index 7ad0836..57293cb 100644
--- a/pkg/common/test/mock/network.go
+++ b/pkg/common/test/mock/network.go
@@ -18,6 +18,7 @@ package mock
 
 import (
 	"bytes"
+	"io"
 	"net"
 	"strings"
 	"time"
@@ -27,6 +28,11 @@ import (
 	"github.com/cloudwego/netpoll"
 )
 
+var (
+	ErrReadTimeout  = errs.New(errs.ErrTimeout, errs.ErrorTypePublic, "read timeout")
+	ErrWriteTimeout = errs.New(errs.ErrTimeout, errs.ErrorTypePublic, "write timeout")
+)
+
 type Conn struct {
 	readTimeout time.Duration
 	zr          network.Reader
@@ -146,7 +152,7 @@ func (m *SlowReadConn) Peek(i int) ([]byte, error) {
 		time.Sleep(100 * time.Millisecond)
 	}
 	if err != nil || len(b) != i {
-		return nil, errs.ErrReadTimeout
+		return nil, ErrReadTimeout
 	}
 	return b, err
 }
@@ -161,6 +167,61 @@ func NewConn(source string) *Conn {
 	}
 }
 
+type BrokenConn struct {
+	*Conn
+}
+
+func (o *BrokenConn) Peek(i int) ([]byte, error) {
+	return nil, io.EOF
+}
+
+func (o *BrokenConn) Flush() error {
+	return errs.ErrConnectionClosed
+}
+
+func NewBrokenConn(source string) *BrokenConn {
+	return &BrokenConn{Conn: NewConn(source)}
+}
+
+type OneTimeConn struct {
+	isRead        bool
+	isFlushed     bool
+	contentLength int
+	*Conn
+}
+
+func (o *OneTimeConn) Peek(n int) ([]byte, error) {
+	if o.isRead {
+		return nil, io.EOF
+	}
+	return o.Conn.Peek(n)
+}
+
+func (o *OneTimeConn) Skip(n int) error {
+	if o.isRead {
+		return io.EOF
+	}
+	o.contentLength -= n
+
+	if o.contentLength == 0 {
+		o.isRead = true
+	}
+
+	return o.Conn.Skip(n)
+}
+
+func (o *OneTimeConn) Flush() error {
+	if o.isFlushed {
+		return errs.ErrConnectionClosed
+	}
+	o.isFlushed = true
+	return o.Conn.Flush()
+}
+
+func NewOneTimeConn(source string) *OneTimeConn {
+	return &OneTimeConn{isRead: false, isFlushed: false, Conn: NewConn(source), contentLength: len(source)}
+}
+
 func NewSlowReadConn(source string) *SlowReadConn {
 	return &SlowReadConn{Conn: NewConn(source)}
 }
@@ -200,7 +261,7 @@ func (m *SlowWriteConn) Flush() error {
 	time.Sleep(100 * time.Millisecond)
 	if err == nil {
 		time.Sleep(m.writeTimeout)
-		return errs.ErrWriteTimeout
+		return ErrWriteTimeout
 	}
 	return err
 }
diff --git a/pkg/common/test/mock/network_test.go b/pkg/common/test/mock/network_test.go
index 2d4ff97..bd6ad5e 100644
--- a/pkg/common/test/mock/network_test.go
+++ b/pkg/common/test/mock/network_test.go
@@ -18,6 +18,7 @@ package mock
 
 import (
 	"context"
+	"io"
 	"testing"
 	"time"
 
@@ -30,6 +31,7 @@ func TestConn(t *testing.T) {
 	t.Run("TestReader", func(t *testing.T) {
 		s1 := "abcdef4343"
 		conn1 := NewConn(s1)
+		assert.Nil(t, conn1.SetWriteTimeout(1))
 		err := conn1.SetReadDeadline(time.Now().Add(time.Millisecond * 100))
 		assert.DeepEqual(t, nil, err)
 		err = conn1.SetReadTimeout(time.Millisecond * 100)
@@ -135,12 +137,16 @@ func TestSlowConn(t *testing.T) {
 	t.Run("TestSlowReadConn", func(t *testing.T) {
 		s1 := "abcdefg"
 		conn := NewSlowReadConn(s1)
+		assert.Nil(t, conn.SetWriteTimeout(1))
+		assert.Nil(t, conn.SetReadTimeout(1))
+		assert.DeepEqual(t, time.Duration(1), conn.readTimeout)
+
 		b, err := conn.Peek(4)
 		assert.DeepEqual(t, nil, err)
 		assert.DeepEqual(t, s1[:4], string(b))
 		conn.Skip(len(s1))
 		_, err = conn.Peek(1)
-		assert.DeepEqual(t, errs.ErrReadTimeout, err)
+		assert.DeepEqual(t, ErrReadTimeout, err)
 		_, err = SlowReadDialer("")
 		assert.DeepEqual(t, nil, err)
 	})
@@ -150,7 +156,7 @@ func TestSlowConn(t *testing.T) {
 		assert.DeepEqual(t, nil, err)
 		conn.SetWriteTimeout(time.Millisecond * 100)
 		err = conn.Flush()
-		assert.DeepEqual(t, errs.ErrWriteTimeout, err)
+		assert.DeepEqual(t, ErrWriteTimeout, err)
 	})
 }
 
@@ -183,3 +189,51 @@ func TestStreamConn(t *testing.T) {
 		})
 	})
 }
+
+func TestBrokenConn_Flush(t *testing.T) {
+	conn := NewBrokenConn("")
+	n, err := conn.Writer().WriteBinary([]byte("Foo"))
+	assert.DeepEqual(t, 3, n)
+	assert.Nil(t, err)
+	assert.DeepEqual(t, errs.ErrConnectionClosed, conn.Flush())
+}
+
+func TestBrokenConn_Peek(t *testing.T) {
+	conn := NewBrokenConn("Foo")
+	buf, err := conn.Peek(3)
+	assert.Nil(t, buf)
+	assert.DeepEqual(t, io.EOF, err)
+}
+
+func TestOneTimeConn_Flush(t *testing.T) {
+	conn := NewOneTimeConn("")
+	n, err := conn.Writer().WriteBinary([]byte("Foo"))
+	assert.DeepEqual(t, 3, n)
+	assert.Nil(t, err)
+	assert.Nil(t, conn.Flush())
+	n, err = conn.Writer().WriteBinary([]byte("Bar"))
+	assert.DeepEqual(t, 3, n)
+	assert.Nil(t, err)
+	assert.DeepEqual(t, errs.ErrConnectionClosed, conn.Flush())
+}
+
+func TestOneTimeConn_Skip(t *testing.T) {
+	conn := NewOneTimeConn("FooBar")
+	buf, err := conn.Peek(3)
+	assert.DeepEqual(t, "Foo", string(buf))
+	assert.Nil(t, err)
+	assert.Nil(t, conn.Skip(3))
+	assert.DeepEqual(t, 3, conn.contentLength)
+
+	buf, err = conn.Peek(3)
+	assert.DeepEqual(t, "Bar", string(buf))
+	assert.Nil(t, err)
+	assert.Nil(t, conn.Skip(3))
+	assert.DeepEqual(t, 0, conn.contentLength)
+
+	buf, err = conn.Peek(3)
+	assert.DeepEqual(t, 0, len(buf))
+	assert.DeepEqual(t, io.EOF, err)
+	assert.DeepEqual(t, io.EOF, conn.Skip(3))
+	assert.DeepEqual(t, 0, conn.contentLength)
+}
diff --git a/pkg/protocol/http1/client_test.go b/pkg/protocol/http1/client_test.go
index 4dd6fcb..7bc389b 100644
--- a/pkg/protocol/http1/client_test.go
+++ b/pkg/protocol/http1/client_test.go
@@ -58,6 +58,7 @@ import (
 	"github.com/cloudwego/hertz/pkg/app/client/retry"
 	"github.com/cloudwego/hertz/pkg/common/config"
 	errs "github.com/cloudwego/hertz/pkg/common/errors"
+	"github.com/cloudwego/hertz/pkg/common/hlog"
 	"github.com/cloudwego/hertz/pkg/common/test/assert"
 	"github.com/cloudwego/hertz/pkg/common/test/mock"
 	"github.com/cloudwego/hertz/pkg/common/utils"
@@ -69,6 +70,8 @@ import (
 	"github.com/cloudwego/netpoll"
 )
 
+var errDialTimeout = errs.New(errs.ErrTimeout, errs.ErrorTypePublic, "dial timeout")
+
 func TestHostClientMaxConnWaitTimeoutWithEarlierDeadline(t *testing.T) {
 	var (
 		emptyBodyCount uint8
@@ -235,7 +238,7 @@ type slowDialer struct {
 
 func (s *slowDialer) DialConnection(network, address string, timeout time.Duration, tlsConfig *tls.Config) (conn network.Conn, err error) {
 	time.Sleep(timeout)
-	return nil, errs.ErrDialTimeout
+	return nil, errDialTimeout
 }
 
 func TestReadTimeoutPriority(t *testing.T) {
@@ -264,7 +267,7 @@ func TestReadTimeoutPriority(t *testing.T) {
 	case <-time.After(time.Second * 2):
 		t.Fatalf("should use readTimeout in request options")
 	case err := <-ch:
-		assert.DeepEqual(t, errs.ErrTimeout.Error(), err.Error())
+		assert.DeepEqual(t, mock.ErrReadTimeout, err)
 	}
 }
 
@@ -334,7 +337,7 @@ func TestWriteTimeoutPriority(t *testing.T) {
 	case <-time.After(time.Second * 2):
 		t.Fatalf("should use writeTimeout in request options")
 	case err := <-ch:
-		assert.DeepEqual(t, errs.ErrWriteTimeout.Error(), err.Error())
+		assert.DeepEqual(t, mock.ErrWriteTimeout, err)
 	}
 }
 
@@ -359,10 +362,10 @@ func TestDialTimeoutPriority(t *testing.T) {
 		ch <- c.Do(context.Background(), req, resp)
 	}()
 	select {
-	case <-time.After(time.Second * 2000):
+	case <-time.After(time.Second * 2):
 		t.Fatalf("should use dialTimeout in request options")
 	case err := <-ch:
-		assert.DeepEqual(t, errs.ErrDialTimeout.Error(), err.Error())
+		assert.DeepEqual(t, errDialTimeout, err)
 	}
 }
 
@@ -479,3 +482,57 @@ type retryConn struct {
 func (w retryConn) SetWriteTimeout(t time.Duration) error {
 	return errors.New("should retry")
 }
+
+func TestConnInPoolRetry(t *testing.T) {
+	c := &HostClient{
+		ClientOptions: &ClientOptions{
+			Dialer: newSlowConnDialer(func(network, addr string) (network.Conn, error) {
+				return mock.NewOneTimeConn("HTTP/1.1 200 OK\r\nContent-Length: 10\r\nContent-Type: foo/bar\r\n\r\n0123456789"), nil
+			}),
+		},
+		Addr: "foobar",
+	}
+
+	req := protocol.AcquireRequest()
+	req.SetRequestURI("http://foobar/baz")
+	req.SetOptions(config.WithWriteTimeout(time.Millisecond * 100))
+	resp := protocol.AcquireResponse()
+
+	logbuf := &bytes.Buffer{}
+	hlog.SetOutput(logbuf)
+
+	err := c.Do(context.Background(), req, resp)
+	assert.Nil(t, err)
+	assert.DeepEqual(t, resp.StatusCode(), 200)
+	assert.DeepEqual(t, string(resp.Body()), "0123456789")
+	assert.True(t, logbuf.String() == "")
+	protocol.ReleaseResponse(resp)
+	resp = protocol.AcquireResponse()
+	err = c.Do(context.Background(), req, resp)
+	assert.Nil(t, err)
+	assert.DeepEqual(t, resp.StatusCode(), 200)
+	assert.DeepEqual(t, string(resp.Body()), "0123456789")
+	assert.True(t, strings.Contains(logbuf.String(), "Client connection attempt times: 1"))
+}
+
+func TestConnNotRetry(t *testing.T) {
+	c := &HostClient{
+		ClientOptions: &ClientOptions{
+			Dialer: newSlowConnDialer(func(network, addr string) (network.Conn, error) {
+				return mock.NewBrokenConn(""), nil
+			}),
+		},
+		Addr: "foobar",
+	}
+
+	req := protocol.AcquireRequest()
+	req.SetRequestURI("http://foobar/baz")
+	req.SetOptions(config.WithWriteTimeout(time.Millisecond * 100))
+	resp := protocol.AcquireResponse()
+	logbuf := &bytes.Buffer{}
+	hlog.SetOutput(logbuf)
+	err := c.Do(context.Background(), req, resp)
+	assert.DeepEqual(t, errs.ErrConnectionClosed, err)
+	assert.True(t, logbuf.String() == "")
+	protocol.ReleaseResponse(resp)
+}
diff --git a/pkg/protocol/http1/client_unix_test.go b/pkg/protocol/http1/client_unix_test.go
index 6e417ba..1e88502 100644
--- a/pkg/protocol/http1/client_unix_test.go
+++ b/pkg/protocol/http1/client_unix_test.go
@@ -35,7 +35,7 @@ import (
 )
 
 func TestGcBodyStream(t *testing.T) {
-	srv := &http.Server{Addr: ":11001", Handler: http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+	srv := &http.Server{Addr: "127.0.0.1:11001", Handler: http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		for range [1024]int{} {
 			w.Write([]byte("hello world\n"))
 		}
@@ -69,7 +69,7 @@ func TestGcBodyStream(t *testing.T) {
 }
 
 func TestMaxConn(t *testing.T) {
-	srv := &http.Server{Addr: ":11002", Handler: http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+	srv := &http.Server{Addr: "127.0.0.1:11002", Handler: http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		w.Write([]byte("hello world\n"))
 	})}
 	go srv.ListenAndServe()