go-omni/transport/http2/soak_test.go

package http2

import (
	"context"
	"crypto/tls"
	"io"
	"net"
	"net/http"
	"os"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	xhttp2 "golang.org/x/net/http2"
)

// TestTLSSoak is short by default so it remains part of normal package
// verification. Linux acceptance runs set GO_OMNI_HTTP2_SOAK_DURATION=30m.
func TestTLSSoak(t *testing.T) {
	duration := 3 * time.Second
	if value := os.Getenv("GO_OMNI_HTTP2_SOAK_DURATION"); value != "" {
		parsed, err := time.ParseDuration(value)
		if err != nil || parsed <= 0 {
			t.Fatalf("invalid GO_OMNI_HTTP2_SOAK_DURATION %q: %v", value, err)
		}
		duration = parsed
	}

	certificate, roots := testCertificate(t)
	server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
		switch request.URL.Path {
		case "/slow":
			timer := time.NewTimer(10 * time.Millisecond)
			defer timer.Stop()
			select {
			case <-timer.C:
			case <-request.Context().Done():
				return
			}
		case "/body":
			if _, err := io.Copy(io.Discard, request.Body); err != nil {
				http.Error(writer, err.Error(), http.StatusBadRequest)
				return
			}
		}
		_, _ = io.WriteString(writer, request.URL.Path)
	}))
	server.MaxConnections = 8
	server.MaxConcurrentRequests = 128
	server.MaxConcurrentStreams = 128
	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.ServeTLS(context.Background(), listener, &tls.Config{
			Certificates: []tls.Certificate{certificate},
			MinVersion:   tls.VersionTLS12,
		})
	}()

	transport := &xhttp2.Transport{
		TLSClientConfig: &tls.Config{
			RootCAs:    roots,
			ServerName: "localhost",
			MinVersion: tls.VersionTLS12,
			NextProtos: []string{"h2"},
		},
		StrictMaxConcurrentStreams: true,
	}
	client := &http.Client{Transport: transport, Timeout: 3 * time.Second}

	var requests, failures atomic.Uint64
	deadline := time.Now().Add(duration)
	var workers sync.WaitGroup
	for worker := 0; worker < 128; worker++ {
		workers.Add(1)
		go func(worker int) {
			defer workers.Done()
			for time.Now().Before(deadline) {
				path := "/health"
				method := http.MethodGet
				var body io.Reader
				if worker%32 == 0 {
					path = "/slow"
				}
				if worker%64 == 0 {
					path = "/body"
					method = http.MethodPost
					body = &repeatingReader{remaining: 64 << 10}
				}
				request, err := http.NewRequestWithContext(context.Background(), method, "https://"+listener.Addr().String()+path, body)
				if err != nil {
					failures.Add(1)
					continue
				}
				response, err := client.Do(request)
				if err != nil {
					failures.Add(1)
					continue
				}
				_, readErr := io.Copy(io.Discard, response.Body)
				response.Body.Close()
				if readErr != nil || response.StatusCode != http.StatusOK {
					failures.Add(1)
					continue
				}
				requests.Add(1)
			}
		}(worker)
	}
	workers.Wait()
	transport.CloseIdleConnections()

	shutdownContext, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	shutdownErr := server.Shutdown(shutdownContext)
	cancel()
	if shutdownErr != nil {
		t.Fatalf("TLS soak shutdown: %v", shutdownErr)
	}
	if err := <-done; err != nil {
		t.Fatalf("TLS soak serve: %v", err)
	}
	if failures.Load() != 0 || requests.Load() == 0 {
		t.Fatalf("TLS soak requests=%d failures=%d stats=%+v", requests.Load(), failures.Load(), server.Stats())
	}
	if stats := server.Stats(); stats.ActiveConnections != 0 || stats.ActiveRequests != 0 {
		t.Fatalf("TLS soak did not reclaim resources: %+v", stats)
	}
}

type repeatingReader struct {
	remaining int
}

func (r *repeatingReader) Read(buffer []byte) (int, error) {
	if r.remaining == 0 {
		return 0, io.EOF
	}
	if len(buffer) > r.remaining {
		buffer = buffer[:r.remaining]
	}
	for index := range buffer {
		buffer[index] = 'x'
	}
	r.remaining -= len(buffer)
	return len(buffer), nil
}

func (r *repeatingReader) Close() error { return nil }