test: cover native slow bodies and pipelining

transport/native/server_test.go

@@ -9,6 +9,7 @@ import (
 	"net"
 	"net/http"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -88,6 +89,85 @@ func TestServerRejectsUnsupportedBodyAndLimits(t *testing.T) {
 	}
 }
 
+func TestServerSlowBodyTimesOutBeforeHandler(t *testing.T) {
+	var handlerCalls atomic.Int32
+	application := omni.NewWithRouter(netrouter.New())
+	application.POST("/body", func(ctx *omni.Context) error {
+		handlerCalls.Add(1)
+		return ctx.OK("ok")
+	})
+	server := native.New(application)
+	server.ReadTimeout = 30 * time.Millisecond
+	server.IdleTimeout = time.Second
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	done := make(chan error, 1)
+	go func() { done <- server.Serve(context.Background(), listener) }()
+	waitForServer(t, listener.Addr().String())
+	defer stopServer(t, server, listener, done)
+
+	conn, err := net.Dial("tcp", listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	conn.SetReadDeadline(time.Now().Add(time.Second))
+	_, _ = io.WriteString(conn, "POST /body HTTP/1.1\r\nHost: test\r\nContent-Length: 4\r\n\r\na")
+	buffer := make([]byte, 1)
+	if _, err := conn.Read(buffer); err == nil {
+		t.Fatal("expected slow request body to time out and close")
+	}
+	if handlerCalls.Load() != 0 {
+		t.Fatal("handler ran before the request body was complete")
+	}
+}
+
+func TestServerPipelinedRequestsAndLargeBody(t *testing.T) {
+	application := omni.NewWithRouter(netrouter.New())
+	application.POST("/body", func(ctx *omni.Context) error {
+		body, err := io.ReadAll(ctx.Request().Body)
+		if err != nil {
+			return err
+		}
+		return ctx.OK(map[string]int{"length": len(body)})
+	})
+	application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") })
+	server, listener, done := startServer(t, application)
+	defer stopServer(t, server, listener, done)
+
+	body := strings.Repeat("x", 64<<10)
+	conn, err := net.Dial("tcp", listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	reader := bufio.NewReader(conn)
+	request := fmt.Sprintf("POST /body HTTP/1.1\r\nHost: test\r\nConnection: keep-alive\r\nContent-Length: %d\r\n\r\n%sGET /health HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n", len(body), body)
+	if _, err := io.WriteString(conn, request); err != nil {
+		t.Fatal(err)
+	}
+	first, err := http.ReadResponse(reader, &http.Request{Method: http.MethodPost})
+	if err != nil {
+		t.Fatal(err)
+	}
+	firstBody, err := io.ReadAll(first.Body)
+	first.Body.Close()
+	if err != nil || first.StatusCode != http.StatusOK || !strings.Contains(string(firstBody), `"length":65536`) {
+		t.Fatalf("unexpected large body response: status=%d body=%s error=%v", first.StatusCode, firstBody, err)
+	}
+	second, err := http.ReadResponse(reader, &http.Request{Method: http.MethodGet})
+	if err != nil {
+		t.Fatal(err)
+	}
+	secondBody, err := io.ReadAll(second.Body)
+	second.Body.Close()
+	if err != nil || second.StatusCode != http.StatusOK || !strings.Contains(string(secondBody), `"ok"`) {
+		t.Fatalf("unexpected pipelined response: status=%d body=%s error=%v", second.StatusCode, secondBody, err)
+	}
+}
+
 func TestServerProtocolBoundaries(t *testing.T) {
 	application := omni.NewWithRouter(netrouter.New())
 	application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") })