smallwebwaf/internal/proxy/timeouts_test.go

package proxy_test

import (
	"errors"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"strconv"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"sneak.berlin/go/smallwebwaf/internal/requestlog"
)

// largeBodySize is more than the connections between the client,
// smallwebwaf and the app can hold while nobody reads, so that a sender
// soon waits.
const largeBodySize = 64 << 20

// writeSize is how much a test sender writes at a time.
const writeSize = 32 << 10

func TestRequestTimeouts(t *testing.T) {
	t.Parallel()

	for _, tc := range []struct {
		name string
		// limit is the setting set to shortTimeout, which runs out; long
		// is one set to longTimeoutSetting, which does not, or "".
		limit, long string
		// appTakesNothing has the app never read, while the client sends
		// as fast as it can; otherwise the app reads, and the client
		// stops sending halfway.
		appTakesNothing bool
		want            int
	}{
		{
			name:  "client request timeout, waiting on the client",
			limit: clientRequestTimeout,
			want:  http.StatusRequestTimeout,
		},
		{
			name:  "upstream request timeout, waiting on the client",
			limit: upstreamRequestTimeout,
			long:  clientRequestTimeout,
			want:  http.StatusRequestTimeout,
		},
		{
			name:            "upstream request timeout, waiting on the app",
			limit:           upstreamRequestTimeout,
			appTakesNothing: true,
			want:            http.StatusGatewayTimeout,
		},
		{
			name:            "client request timeout, waiting on the app",
			limit:           clientRequestTimeout,
			long:            upstreamRequestTimeout,
			appTakesNothing: true,
			want:            http.StatusGatewayTimeout,
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			var (
				app         *httptest.Server
				appURL      string
				sendRequest func(*testing.T, string) net.Conn
				appGotBody  atomic.Bool
			)

			if tc.appTakesNothing {
				appURL, sendRequest = startAppThatTakesNothing(t), sendLargeBody
			} else {
				app = startApp(t, func(_ http.ResponseWriter, r *http.Request) {
					n, _ := io.Copy(io.Discard, r.Body)
					appGotBody.Store(n > 0)
				})
				appURL, sendRequest = app.URL, sendPartOfBody
			}

			env := map[string]string{tc.limit: shortTimeoutSetting, metricsToken: token}
			if tc.long != "" {
				env[tc.long] = longTimeoutSetting
			}

			addr, out := startProxy(t, appURL, env)
			start := time.Now()
			got := readResponse(t, sendRequest(t, addr))
			wantTimedOut(t, start)

			want := tc.want

			if app != nil {
				// Close returns once the app has finished with the request.
				app.Close()

				// Until some of the body has reached the app, smallwebwaf
				// waits on the app, and SPEC.md asks for 504; the timeout
				// runs out then only if the test process is held up.
				if !appGotBody.Load() {
					want = http.StatusGatewayTimeout
				}
			}

			wantStatus(t, got, want)
			wantLine(t, out.requestLine(t), want, requestlog.ActionTimedOut)
			wantLimitHits(t, addr, tc.limit, 1)
		})
	}
}

// startAppThatTakesNothing starts an app that accepts connections and
// never reads from them, and returns its URL.
func startAppThatTakesNothing(t *testing.T) string {
	t.Helper()

	listener, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", localhost+":0")
	if err != nil {
		t.Fatalf("listen: %v", err)
	}

	var (
		mu   sync.Mutex
		held []net.Conn
	)

	hold := func(conn net.Conn) {
		mu.Lock()
		defer mu.Unlock()

		held = append(held, conn)
	}

	go func() {
		for {
			conn, err := listener.Accept()
			if err != nil {
				return
			}

			hold(conn)
		}
	}()

	t.Cleanup(func() {
		_ = listener.Close()

		mu.Lock()
		defer mu.Unlock()

		for _, conn := range held {
			_ = conn.Close()
		}
	})

	return "http://" + listener.Addr().String()
}

// sendPartOfBody sends a request that announces a large body, and only
// the first bytes of it.
func sendPartOfBody(t *testing.T, addr string) net.Conn {
	t.Helper()

	conn := dial(t, addr)
	send(t, conn, "POST /upload HTTP/1.1\r\nHost: app\r\nContent-Length: "+
		strconv.Itoa(largeBodySize)+"\r\n\r\nthe first bytes")

	return conn
}

// sendLargeBody sends a request with a large body, as fast as smallwebwaf
// takes it, from a goroutine of its own.
func sendLargeBody(t *testing.T, addr string) net.Conn {
	t.Helper()

	conn := dial(t, addr)
	send(t, conn, "POST /upload HTTP/1.1\r\nHost: app\r\nContent-Length: "+
		strconv.Itoa(largeBodySize)+"\r\n\r\n")

	go func() {
		chunk := make([]byte, writeSize)
		for range largeBodySize / writeSize {
			_, err := conn.Write(chunk)
			if err != nil {
				return
			}
		}
	}()

	return conn
}

func TestAppTooSlowToAnswer(t *testing.T) {
	t.Parallel()

	app := startApp(t, func(_ http.ResponseWriter, r *http.Request) {
		<-r.Context().Done()
	})
	addr, out := startProxy(t, app.URL, map[string]string{
		upstreamResponseTimeout: shortTimeoutSetting,
		metricsToken:            token,
	})

	start := time.Now()

	wantStatus(t, get(t, addr, "/slow"), http.StatusGatewayTimeout)
	wantTimedOut(t, start)

	line := out.requestLine(t)
	wantLine(t, line, http.StatusGatewayTimeout, requestlog.ActionTimedOut)

	_, answered := line.fields["upstream_status"]
	if answered {
		t.Errorf("log line has upstream_status %v for an app that never answered",
			line.fields["upstream_status"])
	}

	wantLimitHits(t, addr, upstreamResponseTimeout, 1)
}

func TestAppTooSlowToFinishItsAnswer(t *testing.T) {
	t.Parallel()

	app := startApp(t, func(w http.ResponseWriter, r *http.Request) {
		_, _ = io.WriteString(w, "the first part")
		_ = http.NewResponseController(w).Flush()

		<-r.Context().Done()
	})
	addr, out := startProxy(t, app.URL, map[string]string{
		upstreamResponseTimeout: shortTimeoutSetting,
	})

	start := time.Now()
	got := get(t, addr, "/slow")
	wantStatus(t, got, http.StatusOK)

	if string(got.body) != "the first part" || !errors.Is(got.err, io.ErrUnexpectedEOF) {
		t.Errorf("client read %q (%v), want the first part cut off", got.body, got.err)
	}

	wantTimedOut(t, start)

	line := out.requestLine(t)
	wantLine(t, line, http.StatusOK, requestlog.ActionTimedOut)

	if line.UpstreamStatus != http.StatusOK {
		t.Errorf("log line has upstream_status %d, want %d",
			line.UpstreamStatus, http.StatusOK)
	}
}

func TestClientTooSlowToTakeTheAnswer(t *testing.T) {
	t.Parallel()

	app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
		chunk := make([]byte, writeSize)
		for range largeBodySize / writeSize {
			_, err := w.Write(chunk)
			if err != nil {
				return
			}
		}
	})
	addr, out := startProxy(t, app.URL, map[string]string{
		clientResponseTimeout: shortTimeoutSetting,
		metricsToken:          token,
	})

	start := time.Now()

	// The client asks, and never reads the answer.
	conn := dial(t, addr)
	send(t, conn, "GET /large HTTP/1.1\r\nHost: app\r\n\r\n")

	line := out.requestLine(t)
	wantTimedOut(t, start)
	wantLine(t, line, http.StatusOK, requestlog.ActionTimedOut)
	wantLimitHits(t, addr, clientResponseTimeout, 1)
}

func TestClosesAnIdleConnection(t *testing.T) {
	t.Parallel()

	app := startApp(t, func(http.ResponseWriter, *http.Request) {})
	addr, _ := startProxy(t, app.URL, map[string]string{
		clientIdleTimeout: shortTimeoutSetting,
	})

	// The idle time starts once the answer is sent, so after start.
	start := time.Now()
	conn := dial(t, addr)
	send(t, conn, "GET / HTTP/1.1\r\nHost: app\r\n\r\n")
	wantStatus(t, readResponse(t, conn), http.StatusOK)

	// The read deadline readResponse set still bounds this read.
	_, err := conn.Read(make([]byte, 1))
	if !errors.Is(err, io.EOF) {
		t.Fatalf("read on the idle connection: %v, want it closed", err)
	}

	wantTimedOut(t, start)
}