optimize(http1): retry if the error is caused by broken pooled conn

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

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

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.

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
 }

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

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
 }