fix(uploads): bound the silence in an upload body with a re-armed read deadline
A body that stopped arriving held one of the receive slots (#47) and a spool file indefinitely. Every read of the upload body now pushes the connection read deadline UploadIdleTimeout (default 60 s) ahead, so a stalled body fails its read, Accept returns and the slot is released, while a slow but moving body is never cut (ReadTimeout would cut it). Two things found by running it over a socket: - metrics.statusWriter had no Unwrap, so http.ResponseController could not reach the connection and the bound silently did not apply in the production handler stack. It now unwraps; a writer that still cannot set a deadline logs "upload idle bound unavailable". - clearing the deadline on any read error left net/http's pre-reply drain of the unread remainder unbounded, so the connection hung. The deadline is now cleared only on a clean io.EOF. ADR-033 limitation amended. A 1-byte-per-59-s drip is still unbounded. Closes #56 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,114 @@
|
||||
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"
|
||||
"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"
|
||||
)
|
||||
|
||||
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))
|
||||
}
|
||||
Reference in New Issue
Block a user