Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c4233e3e05 | ||
|
|
e946709e82 | ||
|
|
a447523749 | ||
|
|
813e679f93 |
@@ -53,10 +53,12 @@ jobs:
|
|||||||
|
|
||||||
- name: Derive image tags
|
- name: Derive image tags
|
||||||
id: meta
|
id: meta
|
||||||
|
env:
|
||||||
|
# signals#47: a ref name can carry $(...); pass it via env, not ${{ }}.
|
||||||
|
REF: ${{ github.ref }}
|
||||||
run: |
|
run: |
|
||||||
SHA=$(git rev-parse --short HEAD)
|
SHA=$(git rev-parse --short HEAD)
|
||||||
echo "sha-tag=${SHA}" >> "$GITHUB_OUTPUT"
|
echo "sha-tag=${SHA}" >> "$GITHUB_OUTPUT"
|
||||||
REF="${{ github.ref }}"
|
|
||||||
if [[ "$REF" == refs/tags/v* ]]; then
|
if [[ "$REF" == refs/tags/v* ]]; then
|
||||||
echo "version-tag=${REF#refs/tags/}" >> "$GITHUB_OUTPUT"
|
echo "version-tag=${REF#refs/tags/}" >> "$GITHUB_OUTPUT"
|
||||||
fi
|
fi
|
||||||
|
|||||||
@@ -27,6 +27,23 @@ docs it indexes.
|
|||||||
- **Language is Go (ADR-001).** Do not introduce Python or a second language. If you think you
|
- **Language is Go (ADR-001).** Do not introduce Python or a second language. If you think you
|
||||||
need one, that's a new ADR with a real justification, not a default.
|
need one, that's a new ADR with a real justification, not a default.
|
||||||
|
|
||||||
|
## Deploying: a push does not deploy
|
||||||
|
|
||||||
|
Tapir runs on the koala k3s cluster under Flux GitOps. Pushing here only builds the image;
|
||||||
|
`kubectl set image` is reverted by Flux drift correction within seconds. The manifest bump is
|
||||||
|
manual: Flux image-automation in `mathias/infra` covers only `supervisor`, not tapir (infra#111
|
||||||
|
is closed, but `k3s/flux/flux-system/image-automation.yaml` has no tapir policy).
|
||||||
|
|
||||||
|
1. CI (`.gitea/workflows/ci.yml`) builds and pushes `localhost:5000/tapir:<short-sha>` on the
|
||||||
|
main push.
|
||||||
|
2. In `mathias/infra`, bump `image:` in `k3s/apps/tapir/deployment.yaml` to that sha; commit to
|
||||||
|
`main` (infra's own rules and full `task check` apply).
|
||||||
|
3. On koala: `flux reconcile kustomization apps --with-source`, then
|
||||||
|
`kubectl -n tapir rollout status deploy/tapir`. Both need per-action approval in a session.
|
||||||
|
4. Confirm the running image is the new sha and the startup log shows the new code.
|
||||||
|
|
||||||
|
Runbook and why: brain `homelab/failures/flux-reverts-kubectl-set-image-deploy-via-git`.
|
||||||
|
|
||||||
## Things that look reusable but are NOT — read before "reusing"
|
## Things that look reusable but are NOT — read before "reusing"
|
||||||
|
|
||||||
These caused real mistakes that were caught and corrected; the corrections are load-bearing.
|
These caused real mistakes that were caught and corrected; the corrections are load-bearing.
|
||||||
|
|||||||
+13
-4
@@ -1466,10 +1466,19 @@ an hour of Swedish *was* an unknown here; it is now measured, see decision 8.)
|
|||||||
as busy beyond that. Together with the 20-job queue and the running job that bounds the spool
|
as busy beyond that. Together with the 20-job queue and the running job that bounds the spool
|
||||||
at about (4 + 20 + 1) x 200 MiB = 5 GiB of the 6 Gi `emptyDir` (infra): raising any of those or
|
at about (4 + 20 + 1) x 200 MiB = 5 GiB of the 6 Gi `emptyDir` (infra): raising any of those or
|
||||||
`TAPIR_UPLOAD_MAX_BYTES` must revisit that sum (pinned by `TestNewIntake_Defaults...`, not
|
`TAPIR_UPLOAD_MAX_BYTES` must revisit that sum (pinned by `TestNewIntake_Defaults...`, not
|
||||||
enforced against infra). **A slow-drip body holds its slot without a time bound**: the server
|
enforced against infra). **A body is bounded by its silence, not its length (#56, amended
|
||||||
sets `ReadHeaderTimeout` only, because a `ReadTimeout` would cut legitimate 200 MiB uploads on
|
2026-10-04)**: a `ReadTimeout` would cut legitimate 200 MiB uploads on slow links, so instead
|
||||||
slow links; two slow bodies take a user's two slots and two such users make everyone's upload
|
every read of the upload body re-arms the connection's read deadline 60 s ahead
|
||||||
busy. It needs a valid session; recorded as a limitation with a follow-up issue.
|
(`App.UploadIdleTimeout`). A body that stops delivering bytes fails its read, Accept returns
|
||||||
|
and the slot is released; one that keeps moving is never cut. The handler never clears the
|
||||||
|
deadline: an armed deadline also bounds net/http's drain of an unread remainder before it
|
||||||
|
replies, and net/http resets it itself before its background read. A deadline still armed
|
||||||
|
while the file is probed (the multipart part can end before the body's last bytes arrive)
|
||||||
|
costs at most keep-alive reuse of that connection, not the upload. The 60 s default is the
|
||||||
|
deployed value (cmd/tapir does not set it) and is pinned by a test. It needs `http.ResponseController` to reach the connection, so the
|
||||||
|
metrics middleware's writer implements `Unwrap`; where a deadline cannot be set the handler
|
||||||
|
logs `upload idle bound unavailable` rather than silently running unbounded. Still not
|
||||||
|
bounded: a body that drips one byte per 59 s holds its slot for as long as it keeps that up.
|
||||||
|
|
||||||
8. **Chat reuses `chat.Service`; it is not forked.** The uploaded-transcript chat handlers
|
8. **Chat reuses `chat.Service`; it is not forked.** The uploaded-transcript chat handlers
|
||||||
look the row up through the RLS-scoped store, so a non-owner gets 404 by the same mechanism
|
look the row up through the RLS-scoped store, so a non-owner gets 404 by the same mechanism
|
||||||
|
|||||||
@@ -0,0 +1,19 @@
|
|||||||
|
package store_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestMain's Postgres already holds a port. A second instance started in the
|
||||||
|
// same process must still come up, i.e. the port is not derived from something
|
||||||
|
// two starters share (the PID). Observes: startTestPostgres succeeds while
|
||||||
|
// another instance from this process is listening.
|
||||||
|
func TestStartTestPostgres_SecondInstanceStartsWhileFirstHoldsItsPort(t *testing.T) {
|
||||||
|
pg, _, rt, err := startTestPostgres(fmt.Sprintf("tapir-epg-store-second-%d", os.Getpid()))
|
||||||
|
require.NoError(t, err)
|
||||||
|
t.Cleanup(func() { _ = pg.Stop(); _ = os.RemoveAll(rt) })
|
||||||
|
}
|
||||||
@@ -3,6 +3,7 @@ package store_test
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"net"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"testing"
|
"testing"
|
||||||
@@ -25,27 +26,15 @@ var _ ports.Sink = (*store.Store)(nil)
|
|||||||
var dsn string
|
var dsn string
|
||||||
|
|
||||||
func TestMain(m *testing.M) {
|
func TestMain(m *testing.M) {
|
||||||
// Port + runtime/data dirs are per-process (PID-derived) so two concurrent
|
// Runtime/data dirs are per-process (PID-named) and the port is OS-assigned,
|
||||||
// `go test` invocations — e.g. a push-run and a tag-run firing together in CI —
|
// so concurrent `go test` runs on a shared CI host don't collide on either.
|
||||||
// don't collide on a fixed port or a shared data dir (which silently failed
|
// CachePath is shared so the PG archive is downloaded once, not per process.
|
||||||
// both runs). CachePath is shared so the PG archive is downloaded once, not
|
pg, port, rt, err := startTestPostgres(fmt.Sprintf("tapir-epg-store-%d", os.Getpid()))
|
||||||
// per process. Base 54000 keeps this package's range distinct from web's.
|
if err != nil {
|
||||||
port := uint32(54000 + os.Getpid()%1000)
|
|
||||||
dsn = fmt.Sprintf("postgres://postgres:postgres@localhost:%d/postgres?sslmode=disable", port)
|
|
||||||
|
|
||||||
rt := filepath.Join(os.TempDir(), fmt.Sprintf("tapir-epg-store-%d", os.Getpid()))
|
|
||||||
pg := embeddedpostgres.NewDatabase(
|
|
||||||
embeddedpostgres.DefaultConfig().
|
|
||||||
Port(port).
|
|
||||||
RuntimePath(rt).
|
|
||||||
DataPath(filepath.Join(rt, "data")).
|
|
||||||
BinariesPath(filepath.Join(rt, "bin")).
|
|
||||||
CachePath(filepath.Join(os.TempDir(), "tapir-epg-cache")),
|
|
||||||
)
|
|
||||||
if err := pg.Start(); err != nil {
|
|
||||||
fmt.Fprintf(os.Stderr, "embedded-postgres start: %v\n", err)
|
fmt.Fprintf(os.Stderr, "embedded-postgres start: %v\n", err)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
|
dsn = fmt.Sprintf("postgres://postgres:postgres@localhost:%d/postgres?sslmode=disable", port)
|
||||||
|
|
||||||
code := m.Run()
|
code := m.Run()
|
||||||
|
|
||||||
@@ -56,6 +45,44 @@ func TestMain(m *testing.M) {
|
|||||||
os.Exit(code)
|
os.Exit(code)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// freePort asks the OS for an unused TCP port. The port is released before
|
||||||
|
// return, so another process can grab it first; startTestPostgres retries.
|
||||||
|
func freePort() (uint32, error) {
|
||||||
|
l, err := net.Listen("tcp", "127.0.0.1:0")
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("listen for free port: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = l.Close() }()
|
||||||
|
return uint32(l.Addr().(*net.TCPAddr).Port), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// startTestPostgres starts an embedded Postgres on an OS-assigned free port and
|
||||||
|
// returns it with that port and its runtime dir (caller removes it). A port
|
||||||
|
// taken between freePort and Start (shared CI host) is retried with a new one.
|
||||||
|
func startTestPostgres(name string) (*embeddedpostgres.EmbeddedPostgres, uint32, string, error) {
|
||||||
|
rt := filepath.Join(os.TempDir(), name)
|
||||||
|
var lastErr error
|
||||||
|
for range 3 {
|
||||||
|
port, err := freePort()
|
||||||
|
if err != nil {
|
||||||
|
return nil, 0, rt, err
|
||||||
|
}
|
||||||
|
pg := embeddedpostgres.NewDatabase(
|
||||||
|
embeddedpostgres.DefaultConfig().
|
||||||
|
Port(port).
|
||||||
|
RuntimePath(rt).
|
||||||
|
DataPath(filepath.Join(rt, "data")).
|
||||||
|
BinariesPath(filepath.Join(rt, "bin")).
|
||||||
|
CachePath(filepath.Join(os.TempDir(), "tapir-epg-cache")),
|
||||||
|
)
|
||||||
|
if lastErr = pg.Start(); lastErr == nil {
|
||||||
|
return pg, port, rt, nil
|
||||||
|
}
|
||||||
|
lastErr = fmt.Errorf("start on port %d: %w", port, lastErr)
|
||||||
|
}
|
||||||
|
return nil, 0, rt, lastErr
|
||||||
|
}
|
||||||
|
|
||||||
// uuids — fixed so tests are deterministic. user_id/video_id are UUID columns.
|
// uuids — fixed so tests are deterministic. user_id/video_id are UUID columns.
|
||||||
const (
|
const (
|
||||||
userA = "11111111-1111-1111-1111-111111111111"
|
userA = "11111111-1111-1111-1111-111111111111"
|
||||||
|
|||||||
@@ -125,6 +125,11 @@ type statusWriter struct {
|
|||||||
wroteHeader bool
|
wroteHeader bool
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Unwrap exposes the underlying writer to http.ResponseController, so a handler behind this
|
||||||
|
// middleware can still set a connection read deadline (#56: without it the upload idle bound
|
||||||
|
// was silently unavailable in production).
|
||||||
|
func (s *statusWriter) Unwrap() http.ResponseWriter { return s.ResponseWriter }
|
||||||
|
|
||||||
func (s *statusWriter) WriteHeader(code int) {
|
func (s *statusWriter) WriteHeader(code int) {
|
||||||
if !s.wroteHeader {
|
if !s.wroteHeader {
|
||||||
s.code = code
|
s.code = code
|
||||||
|
|||||||
@@ -0,0 +1,11 @@
|
|||||||
|
package web
|
||||||
|
|
||||||
|
import "time"
|
||||||
|
|
||||||
|
// DefaultUploadIdleTimeout exposes the production idle bound (#56) to the black-box tests:
|
||||||
|
// cmd/tapir never sets App.UploadIdleTimeout, so this constant IS the deployed value.
|
||||||
|
const DefaultUploadIdleTimeout = defaultUploadIdleTimeout
|
||||||
|
|
||||||
|
// UploadIdleTimeoutForTest exposes the bound an App resolves to (#56), so a test can pin the
|
||||||
|
// value the zero-value App -- production's -- actually uses rather than only the constant.
|
||||||
|
func (a *App) UploadIdleTimeoutForTest() time.Duration { return a.uploadIdleTimeout() }
|
||||||
@@ -116,6 +116,11 @@ type App struct {
|
|||||||
// front rather than discovered by failing.
|
// front rather than discovered by failing.
|
||||||
Uploads Uploader
|
Uploads Uploader
|
||||||
UploadLimits uploads.Limits
|
UploadLimits uploads.Limits
|
||||||
|
// UploadIdleTimeout is how long an upload body may go without delivering a byte
|
||||||
|
// before the connection read is cut off and its receive slot freed (#56). Zero
|
||||||
|
// means defaultUploadIdleTimeout. It bounds silence, not the whole body, so a
|
||||||
|
// large upload on a slow but moving link still completes.
|
||||||
|
UploadIdleTimeout time.Duration
|
||||||
// UploadChatMaxChars is the transcript budget for chat over an uploaded
|
// UploadChatMaxChars is the transcript budget for chat over an uploaded
|
||||||
// recording. The caption default (18,000 chars, ~22 minutes of speech) would
|
// recording. The caption default (18,000 chars, ~22 minutes of speech) would
|
||||||
// keep only the first third of an hour-long meeting.
|
// keep only the first third of an hour-long meeting.
|
||||||
|
|||||||
+48
-1
@@ -84,7 +84,7 @@ func (a *App) handleUploadPost(w http.ResponseWriter, r *http.Request) {
|
|||||||
if !ok {
|
if !ok {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
r.Body = http.MaxBytesReader(w, r.Body, a.UploadLimits.MaxBytes+uploadOverhead)
|
r.Body = http.MaxBytesReader(w, a.idleBoundedBody(w, r.Body), a.UploadLimits.MaxBytes+uploadOverhead)
|
||||||
mr, err := r.MultipartReader()
|
mr, err := r.MultipartReader()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "Choose an audio file to upload.")
|
a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "Choose an audio file to upload.")
|
||||||
@@ -158,6 +158,53 @@ func (a *App) handleUploadPost(w http.ResponseWriter, r *http.Request) {
|
|||||||
a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "Choose an audio file to upload.")
|
a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "Choose an audio file to upload.")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// defaultUploadIdleTimeout is how long an upload body may deliver nothing before its
|
||||||
|
// connection read is cut off (#56). A stall this long is not a slow link but a dead or
|
||||||
|
// hostile one; a legitimate upload, however slow, keeps delivering bytes.
|
||||||
|
const defaultUploadIdleTimeout = 60 * time.Second
|
||||||
|
|
||||||
|
// uploadIdleTimeout is the idle bound this App applies: UploadIdleTimeout when set, else
|
||||||
|
// defaultUploadIdleTimeout. cmd/tapir leaves it unset, so production runs the default.
|
||||||
|
func (a *App) uploadIdleTimeout() time.Duration {
|
||||||
|
if a.UploadIdleTimeout <= 0 {
|
||||||
|
return defaultUploadIdleTimeout
|
||||||
|
}
|
||||||
|
return a.UploadIdleTimeout
|
||||||
|
}
|
||||||
|
|
||||||
|
// idleBoundedBody bounds the SILENCE in an upload body, not its total duration (#56): every
|
||||||
|
// read first pushes the connection's read deadline idle into the future, so a body that
|
||||||
|
// keeps moving is never cut, and one that stops fails its read with a timeout. That read
|
||||||
|
// error reaches Accept as a client read error, so Accept returns and its receive slot is
|
||||||
|
// released. A whole-request ReadTimeout would instead cut a large upload on a slow link.
|
||||||
|
//
|
||||||
|
// If the writer cannot set a read deadline (a wrapper without Unwrap, or a test recorder)
|
||||||
|
// the body is returned unbounded and that is logged: a control that silently does not
|
||||||
|
// apply is worse than one that says so.
|
||||||
|
func (a *App) idleBoundedBody(w http.ResponseWriter, body io.ReadCloser) io.ReadCloser {
|
||||||
|
idle := a.uploadIdleTimeout()
|
||||||
|
rc := http.NewResponseController(w)
|
||||||
|
if err := rc.SetReadDeadline(time.Now().Add(idle)); err != nil {
|
||||||
|
a.logger().Warn("upload idle bound unavailable", "err", err)
|
||||||
|
return body
|
||||||
|
}
|
||||||
|
return &idleDeadlineBody{ReadCloser: body, rc: rc, idle: idle}
|
||||||
|
}
|
||||||
|
|
||||||
|
type idleDeadlineBody struct {
|
||||||
|
io.ReadCloser
|
||||||
|
rc *http.ResponseController
|
||||||
|
idle time.Duration
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *idleDeadlineBody) Read(p []byte) (int, error) {
|
||||||
|
_ = b.rc.SetReadDeadline(time.Now().Add(b.idle))
|
||||||
|
// Never cleared here: net/http clears the read deadline itself before its background
|
||||||
|
// read once the handler has the body, and an armed deadline after an error also bounds
|
||||||
|
// net/http's drain of an unread remainder before it replies.
|
||||||
|
return b.ReadCloser.Read(p)
|
||||||
|
}
|
||||||
|
|
||||||
// abandoned reports, logs at INFO and answers 499, because the client has already gone: its request
|
// abandoned reports, logs at INFO and answers 499, because the client has already gone: its request
|
||||||
// context is cancelled, so there is nobody to answer and rendering a page would run its
|
// context is cancelled, so there is nobody to answer and rendering a page would run its
|
||||||
// database read on a dead context and log a false ERROR. A hang-up mid-upload is routine.
|
// database read on a dead context and log a false ERROR. A hang-up mid-upload is routine.
|
||||||
|
|||||||
@@ -0,0 +1,161 @@
|
|||||||
|
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)
|
||||||
|
}
|
||||||
|
|
||||||
|
// The bound an App actually resolves to, not the constant: production's App leaves
|
||||||
|
// UploadIdleTimeout unset, so the zero value must resolve to exactly 60 s, and a set value
|
||||||
|
// must be used as given.
|
||||||
|
func TestAppUploadIdleTimeout_ResolvesUnsetToSixtySecondsAndHonoursASetValue(t *testing.T) {
|
||||||
|
require.Equal(t, 60*time.Second, (&web.App{}).UploadIdleTimeoutForTest())
|
||||||
|
require.Equal(t, 60*time.Second, (&web.App{UploadIdleTimeout: -time.Second}).UploadIdleTimeoutForTest())
|
||||||
|
require.Equal(t, 300*time.Millisecond, (&web.App{UploadIdleTimeout: 300 * time.Millisecond}).UploadIdleTimeoutForTest())
|
||||||
|
}
|
||||||
|
|
||||||
|
// With UploadIdleTimeout unset, a body with 100 ms gaps between reads is not cut (a resolved
|
||||||
|
// bound in milliseconds or nanoseconds would cut it). The exact value is pinned above.
|
||||||
|
func TestUploadPost_AnUnsetIdleTimeoutDoesNotCutABodyWithSubSecondGaps(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")
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user