tapir/internal/web/uploads_idle_handler_test.go

package web_test

// #56: a request body that stops arriving must not hold a receive slot forever. The bound is a
// read deadline on the connection, re-armed on every read, so it only exercises over a real
// socket and through the production handler stack (metrics.HTTPMiddleware wraps the writer).

import (
	"bufio"
	"bytes"
	"errors"
	"fmt"
	"log/slog"
	"mime/multipart"
	"net"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"git.d-ma.be/mathias/tapir/internal/metrics"
	"git.d-ma.be/mathias/tapir/internal/uploads"
	"git.d-ma.be/mathias/tapir/internal/web"
)

func multipartBody(t *testing.T, content []byte) (body []byte, contentType string) {
	t.Helper()
	var buf bytes.Buffer
	mw := multipart.NewWriter(&buf)
	fw, err := mw.CreateFormFile("file", "a.m4a")
	require.NoError(t, err)
	_, err = fw.Write(content)
	require.NoError(t, err)
	require.NoError(t, mw.Close())
	return buf.Bytes(), mw.FormDataContentType()
}

// rawUpload opens a connection and sends the request headers; the caller feeds the body.
func rawUpload(t *testing.T, addr, contentType string, length int) net.Conn {
	t.Helper()
	c, err := net.Dial("tcp", addr)
	require.NoError(t, err)
	t.Cleanup(func() { _ = c.Close() })
	_, err = fmt.Fprintf(c, "POST /uploads HTTP/1.1\r\nHost: x\r\nContent-Type: %s\r\nContent-Length: %d\r\n\r\n", contentType, length)
	require.NoError(t, err)
	return c
}

// serverEnded reports whether the server ended the request (answered it or closed the
// connection) before wait elapsed, rather than leaving the client waiting.
func serverEnded(t *testing.T, c net.Conn, wait time.Duration) bool {
	t.Helper()
	require.NoError(t, c.SetReadDeadline(time.Now().Add(wait)))
	_, err := http.ReadResponse(bufio.NewReader(c), nil)
	var ne net.Error
	return !errors.As(err, &ne) || !ne.Timeout()
}

// readStatus returns the response status code, or 0 if none arrived before wait elapsed.
func readStatus(t *testing.T, c net.Conn, wait time.Duration) int {
	t.Helper()
	require.NoError(t, c.SetReadDeadline(time.Now().Add(wait)))
	resp, err := http.ReadResponse(bufio.NewReader(c), nil)
	if err != nil {
		return 0
	}
	_ = resp.Body.Close()
	return resp.StatusCode
}

func newIdleBoundServer(t *testing.T, idle time.Duration) (addr string) {
	t.Helper()
	app, _ := newUploadApp(t, audioOK, 200<<20)
	app.UploadIdleTimeout = idle
	app.Uploads.(*uploads.Intake).MaxReceivesPerUser = 1
	srv := httptest.NewServer(metrics.HTTPMiddleware(app.Router()))
	t.Cleanup(srv.Close)
	return strings.TrimPrefix(srv.URL, "http://")
}

func TestUploadPost_AConnectionReadDeadlineEndsABodyThatStallsAndReleasesItsReceiveSlot(t *testing.T) {
	addr := newIdleBoundServer(t, 200*time.Millisecond)
	body, ct := multipartBody(t, bytes.Repeat([]byte("a"), 8000))

	stalled := rawUpload(t, addr, ct, len(body))
	_, err := stalled.Write(body[:4000]) // half the body, then nothing more
	require.NoError(t, err)

	require.True(t, serverEnded(t, stalled, 3*time.Second),
		"the server never ended the stalled request: it still holds its slot")

	next := rawUpload(t, addr, ct, len(body))
	_, err = next.Write(body)
	require.NoError(t, err)
	require.Equal(t, http.StatusSeeOther, readStatus(t, next, 3*time.Second),
		"the same user's next upload gets the slot the stalled one held")
}

func TestUploadPost_TheReadDeadlineIsReArmedOnEveryReadSoASlowButMovingBodyIsAccepted(t *testing.T) {
	addr := newIdleBoundServer(t, 300*time.Millisecond)
	body, ct := multipartBody(t, bytes.Repeat([]byte("a"), 2000))

	c := rawUpload(t, addr, ct, len(body))
	// ~1.2 s in total, four times the idle bound, but never 300 ms without a byte.
	const chunks = 20
	step := len(body)/chunks + 1
	for i := 0; i < len(body); i += step {
		_, err := c.Write(body[i:min(i+step, len(body))])
		require.NoError(t, err)
		time.Sleep(60 * time.Millisecond)
	}

	require.Equal(t, http.StatusSeeOther, readStatus(t, c, 3*time.Second))
}

// cmd/tapir never sets UploadIdleTimeout, so the default is the value production runs.
// Shorter and legitimate uploads on a slow link are cut; much longer and the bound is moot.
func TestDefaultUploadIdleTimeoutIsSixtySeconds(t *testing.T) {
	require.Equal(t, 60*time.Second, web.DefaultUploadIdleTimeout)
}

// With UploadIdleTimeout unset the default applies: a body with 100 ms gaps between reads is
// accepted (a default in milliseconds or nanoseconds would cut it).
func TestUploadPost_AnUnsetIdleTimeoutAppliesTheDefaultAndAcceptsABodyWithSubSecondGaps(t *testing.T) {
	addr := newIdleBoundServer(t, 0)
	body, ct := multipartBody(t, bytes.Repeat([]byte("a"), 1000))

	c := rawUpload(t, addr, ct, len(body))
	const chunks = 5
	step := len(body)/chunks + 1
	for i := 0; i < len(body); i += step {
		time.Sleep(100 * time.Millisecond)
		_, err := c.Write(body[i:min(i+step, len(body))])
		require.NoError(t, err)
	}

	require.Equal(t, http.StatusSeeOther, readStatus(t, c, 3*time.Second))
}

// A writer that cannot set a read deadline leaves the body unbounded; that must be said in
// the log, not happen silently. The in-process recorder is such a writer.
func TestUploadPost_AWriterWithoutReadDeadlineSupportLogsThatTheIdleBoundIsUnavailable(t *testing.T) {
	app, _ := newUploadApp(t, audioOK, 200<<20)
	buf := &lockedBuf{}
	app.Log = slog.New(slog.NewTextHandler(buf, nil))

	do(t, app, uploadReq(t, "t", "a.m4a", []byte("x")))

	require.Contains(t, buf.String(), "upload idle bound unavailable")
}