test: exercise native connection churn

transport/native/server_test.go

@@ -9,6 +9,7 @@ import (
 	"net"
 	"net/http"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -64,7 +65,6 @@ func TestServerRejectsUnsupportedBodyAndLimits(t *testing.T) {
 	}
 	done := make(chan error, 1)
 	go func() { done <- server.Serve(context.Background(), listener) }()
-	waitForServer(t, listener.Addr().String())
 	defer stopServer(t, server, listener, done)
 
 	for _, requestText := range []string{
@@ -462,7 +462,6 @@ func TestServerConnectionLimitReleasesAfterRequest(t *testing.T) {
 	}
 	done := make(chan error, 1)
 	go func() { done <- server.Serve(context.Background(), listener) }()
-	waitForServer(t, listener.Addr().String())
 	defer stopServer(t, server, listener, done)
 
 	first, err := net.Dial("tcp", listener.Addr().String())
@@ -512,6 +511,61 @@ func TestServerConnectionLimitReleasesAfterRequest(t *testing.T) {
 	}
 }
 
+func TestServerConnectionChurnReusesStateSafely(t *testing.T) {
+	application := omni.NewWithRouter(netrouter.New())
+	application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") })
+	server := native.New(application)
+	server.MaxConnections = 16
+	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)
+
+	const workers = 8
+	const requestsPerWorker = 100
+	errors := make(chan error, workers)
+	var group sync.WaitGroup
+	for worker := 0; worker < workers; worker++ {
+		group.Add(1)
+		go func() {
+			defer group.Done()
+			for index := 0; index < requestsPerWorker; index++ {
+				conn, err := net.Dial("tcp", listener.Addr().String())
+				if err != nil {
+					errors <- err
+					return
+				}
+				_, writeErr := io.WriteString(conn, "GET /health HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n")
+				if writeErr == nil {
+					response, readErr := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: http.MethodGet})
+					if readErr != nil {
+						writeErr = readErr
+					} else {
+						response.Body.Close()
+						if response.StatusCode != http.StatusOK {
+							writeErr = fmt.Errorf("unexpected status %d", response.StatusCode)
+						}
+					}
+				}
+				conn.Close()
+				if writeErr != nil {
+					errors <- writeErr
+					return
+				}
+			}
+		}()
+	}
+	group.Wait()
+	close(errors)
+	for err := range errors {
+		t.Fatal(err)
+	}
+}
+
 func TestServerReadHeaderTimeout(t *testing.T) {
 	application := omni.NewWithRouter(netrouter.New())
 	application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") })