Compare commits

..
Author SHA1 Message Date
mathiasandClaude Sonnet 5.5 5baffd0189 fix(test): strip colour codes before looking for the lint skip notice (CI run 8937)
CI / Lint / Test / Vet (pull_request) Successful in 30s
CI / Build & Import (pull_request) Skipped
CI / Deploy via GitOps (pull_request) Skipped
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-03 21:18:05 +02:00
mathiasandClaude Sonnet 5.5 228bcc77da test(lint): the skip notice is found behind CI's colour codes, red
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-03 21:18:01 +02:00
mathiasandClaude Sonnet 5.5 4ef446548d docs(adr): spool arithmetic and the unbounded slow-drip limitation for the receive cap (#47)
CI / Lint / Test / Vet (pull_request) Failing after 30s
CI / Build & Import (pull_request) Skipped
CI / Deploy via GitOps (pull_request) Skipped
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-03 21:12:48 +02:00
mathiasandClaude Sonnet 5.5 145949875e fix(audio): keep a bounded head of ffmpeg's output instead of buffering all of it (#49)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-03 21:12:48 +02:00
mathiasandClaude Sonnet 5.5 82aeac7c19 fix(store): FailUpload only fails a queued or transcribing upload (review of #55)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-03 21:12:48 +02:00
mathiasandClaude Sonnet 5.5 1b4747ed03 test: FailUpload must not overwrite a finished upload; ffmpeg stderr is bounded, red (review of #55)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-03 21:12:34 +02:00
mathiasandClaude Sonnet 5.5 9939afb90c test: pin the production receive limits, split the vacuous drain test, harden the lint-gate test (review of #55)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-03 21:12:34 +02:00
8 changed files with 190 additions and 19 deletions
+9
View File
@@ -1457,6 +1457,15 @@ an hour of Swedish *was* an unknown here; it is now measured, see decision 8.)
perhaps one time in five at current volume). Jobs still queued at shutdown stay `queued` until perhaps one time in five at current volume). Jobs still queued at shutdown stay `queued` until
the next start's sweep. The user re-uploads, with an honest message. the next start's sweep. The user re-uploads, with an honest message.
*Receive limits (#47).* Request bodies are received at most 4 at a time, 2 per user, answered
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
`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
sets `ReadHeaderTimeout` only, because a `ReadTimeout` would cut legitimate 200 MiB uploads on
slow links; two slow bodies take a user's two slots and two such users make everyone's upload
busy. It needs a valid session; recorded as a limitation with a follow-up issue.
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
as ADR-027's summary gate, **before any model is called**. ADR-027's decisions (ephemeral, as ADR-027's summary gate, **before any model is called**. ADR-027's decisions (ephemeral,
+40 -11
View File
@@ -5,6 +5,7 @@ import (
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"os" "os"
"sync"
"syscall" "syscall"
"testing" "testing"
"time" "time"
@@ -32,9 +33,20 @@ func TestShutdownContext_IsCancelledBySIGTERMAndSIGINT(t *testing.T) {
} }
// serveHTTP must not return while a request is in flight or the worker is still recording // serveHTTP must not return while a request is in flight or the worker is still recording
// its job: ListenAndServe returns ErrServerClosed the moment Shutdown starts, so returning // its job: Serve returns http.ErrServerClosed as soon as Shutdown STARTS, and returning
// from it let the process exit before either finished (#50). // from it let the process exit before either finished (#50). Each half is tested with the
func TestServeHTTP_ReturnsOnlyAfterTheInFlightRequestAndTheWorkerHaveFinished(t *testing.T) { // other already satisfied, so neither can hold the call open on the other's behalf.
func pendingFor(t *testing.T, returned <-chan error, why string) {
t.Helper()
select {
case err := <-returned:
t.Fatalf("serveHTTP returned (%v) %s", err, why)
case <-time.After(150 * time.Millisecond):
}
}
func TestServeHTTP_DrainsAnInFlightRequestBeforeReturning(t *testing.T) {
release := make(chan struct{}) release := make(chan struct{})
entered := make(chan struct{}) entered := make(chan struct{})
srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -44,8 +56,12 @@ func TestServeHTTP_ReturnsOnlyAfterTheInFlightRequestAndTheWorkerHaveFinished(t
})) }))
srv.Start() srv.Start()
defer srv.Close() defer srv.Close()
var once sync.Once
releaseHandler := func() { once.Do(func() { close(release) }) }
defer releaseHandler() // runs before srv.Close: a failing test must not hang on the blocked handler
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
workerDone := make(chan struct{}) workerDone := make(chan struct{})
close(workerDone) // the worker is already done: only the request can hold serveHTTP open
returned := make(chan error, 1) returned := make(chan error, 1)
go func() { returned <- serveHTTP(ctx, srv.Config, srv.Listener, nil, workerDone, 5*time.Second) }() go func() { returned <- serveHTTP(ctx, srv.Config, srv.Listener, nil, workerDone, 5*time.Second) }()
@@ -53,17 +69,30 @@ func TestServeHTTP_ReturnsOnlyAfterTheInFlightRequestAndTheWorkerHaveFinished(t
<-entered <-entered
cancel() cancel()
assertPending := func(why string) { pendingFor(t, returned, "while a request was still in flight")
t.Helper() releaseHandler()
select { select {
case err := <-returned: case err := <-returned:
t.Fatalf("serveHTTP returned (%v) %s", err, why) if err != nil {
case <-time.After(150 * time.Millisecond): t.Fatalf("serveHTTP: %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("serveHTTP did not return once the request finished")
} }
} }
assertPending("while a request was still in flight")
close(release) func TestServeHTTP_WaitsForTheWorkerBeforeReturning(t *testing.T) {
assertPending("before the worker finished") srv := httptest.NewUnstartedServer(http.NotFoundHandler())
srv.Start()
defer srv.Close()
ctx, cancel := context.WithCancel(context.Background())
workerDone := make(chan struct{})
returned := make(chan error, 1)
go func() { returned <- serveHTTP(ctx, srv.Config, srv.Listener, nil, workerDone, 5*time.Second) }() // no request in flight
cancel()
pendingFor(t, returned, "before the worker finished")
close(workerDone) close(workerDone)
select { select {
case err := <-returned: case err := <-returned:
@@ -71,7 +100,7 @@ func TestServeHTTP_ReturnsOnlyAfterTheInFlightRequestAndTheWorkerHaveFinished(t
t.Fatalf("serveHTTP: %v", err) t.Fatalf("serveHTTP: %v", err)
} }
case <-time.After(2 * time.Second): case <-time.After(2 * time.Second):
t.Fatal("serveHTTP did not return once the request and the worker had finished") t.Fatal("serveHTTP did not return once the worker finished")
} }
} }
+6 -2
View File
@@ -47,11 +47,15 @@ func (f *FFmpeg) ToWAV(ctx context.Context, src, dst string) error {
args = append(args, "-f", "wav", dst) args = append(args, "-f", "wav", dst)
cmd := exec.CommandContext(ctx, bin, args...) cmd := exec.CommandContext(ctx, bin, args...)
cmd.WaitDelay = killGrace cmd.WaitDelay = killGrace
if out, err := cmd.CombinedOutput(); err != nil { // Only a bounded head of ffmpeg's output is kept: a hostile file made it print ~0.4
// bytes per input byte, all of which CombinedOutput would buffer and copy (#49).
out := &headWriter{max: maxStderr}
cmd.Stdout, cmd.Stderr = out, out
if err := cmd.Run(); err != nil {
if ctxErr := ctx.Err(); ctxErr != nil { if ctxErr := ctx.Err(); ctxErr != nil {
return fmt.Errorf("audio: ffmpeg: %w", ctxErr) return fmt.Errorf("audio: ffmpeg: %w", ctxErr)
} }
return fmt.Errorf("audio: ffmpeg: %w: %s", err, strings.TrimSpace(string(out))) return fmt.Errorf("audio: ffmpeg: %w: %s", err, strings.TrimSpace(out.buf.String()))
} }
// A zero exit that wrote nothing is still a failure: never hand whisper a // A zero exit that wrote nothing is still a failure: never hand whisper a
// missing file and let the error surface three layers later. // missing file and let the error surface three layers later.
+12
View File
@@ -126,3 +126,15 @@ printf data > "$last"`)
assert.Equal(t, "32", args[ms+1]) assert.Equal(t, "32", args[ms+1])
assert.Less(t, ms, in, "an input option placed after -i is applied to the output") assert.Less(t, ms, in, "an input option placed after -i is applied to the output")
} }
func TestFFmpegToWAV_ABoundedPartOfAFloodingStderrIsKept(t *testing.T) {
// A hostile mp3 made ffmpeg print ~0.4 bytes of stderr per input byte (83 MB for a
// 209 MB file), all of it buffered, then copied twice into the error (#49 review).
bin := stub(t, `head -c 5000000 /dev/zero | tr '\0' 'e' >&2; exit 1`)
err := (&audio.FFmpeg{Bin: bin}).ToWAV(context.Background(), "/spool/in", filepath.Join(t.TempDir(), "o.wav"))
require.Error(t, err)
assert.Less(t, len(err.Error()), 16*1024, "the error carries a bounded head of stderr, not all of it")
assert.Contains(t, err.Error(), "eeee", "and still carries some of it")
}
@@ -88,12 +88,16 @@ func (s *Store) Heartbeat(ctx context.Context, userID, id string) error {
WHERE id = $1 AND user_id = $2`) WHERE id = $1 AND user_id = $2`)
} }
// FailUpload moves an upload to failed with a typed reason. // FailUpload moves a queued or transcribing upload to failed with a typed reason. A
// finished upload is left as it was: a shutdown can record a job as failed after its
// transcript was already committed (the commit lands, the caller sees a cancelled
// context), and that must not hide the text. Such a call touches no row and returns
// ErrUploadNotFound.
func (s *Store) FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error { func (s *Store) FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error {
return s.updateUpload(ctx, userID, id, "fail upload", return s.updateUpload(ctx, userID, id, "fail upload",
`UPDATE uploaded_transcripts `UPDATE uploaded_transcripts
SET status = 'failed', error_kind = $3, finished_at = NOW() SET status = 'failed', error_kind = $3, finished_at = NOW()
WHERE id = $1 AND user_id = $2`, WHERE id = $1 AND user_id = $2 AND status IN ('queued', 'transcribing')`,
string(kind)) string(kind))
} }
@@ -0,0 +1,47 @@
package store_test
// A shutdown can record a job as failed after its transcript was already committed (the
// commit lands, the caller sees a cancelled context). FailUpload must not be able to turn a
// finished row into a failed one and hide the text (#50 review).
import (
"context"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"git.d-ma.be/mathias/tapir/internal/domain"
)
func TestFailUpload_LeavesAFinishedUploadAsItWas(t *testing.T) {
s := newStore(t)
ctx := context.Background()
id := startedUpload(t, s)
require.NoError(t, s.StartTranscribing(ctx, userA, id))
require.NoError(t, s.CompleteUpload(ctx, userA, id, domain.Transcription{Text: "synthetic text", Endpoint: "iguana/kb-whisper"}))
_ = s.FailUpload(ctx, userA, id, domain.UploadErrInterrupted)
got, err := s.GetUpload(ctx, userA, id)
require.NoError(t, err)
assert.Equal(t, domain.UploadDone, got.Status)
assert.Equal(t, "synthetic text", got.Content)
assert.Empty(t, got.ErrorKind)
}
func TestFailUpload_StillFailsAQueuedOrTranscribingUpload(t *testing.T) {
s := newStore(t)
ctx := context.Background()
queued := startedUpload(t, s)
running, err := s.CreateUpload(ctx, userA, "r", false)
require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, running))
for _, id := range []string{queued, running} {
require.NoError(t, s.FailUpload(ctx, userA, id, domain.UploadErrBusy))
got, err := s.GetUpload(ctx, userA, id)
require.NoError(t, err)
assert.Equal(t, domain.UploadFailed, got.Status)
}
}
+31
View File
@@ -0,0 +1,31 @@
package uploads_test
// The limits production runs on are the DEFAULTS: nothing outside tests sets
// MaxReceives or MaxReceivesPerUser. Pinned here with the literal numbers, so changing a
// default is a visible decision (#47 review). With 4 receives, 20 queued and 1 running at
// the 200 MiB cap, the spool needs ~5 GiB of the 6 Gi emptyDir (infra); raising any of
// them or TAPIR_UPLOAD_MAX_BYTES must revisit that arithmetic.
import (
"context"
"strings"
"testing"
"time"
"git.d-ma.be/mathias/tapir/internal/uploads"
)
func TestNewIntake_DefaultsAllowTwoReceivesPerUserAndFourOverall(t *testing.T) {
r := newRig(t)
in := uploads.NewIntake(r.st, r.pr, r.sub, uploads.Limits{MaxBytes: 1024, MaxDuration: time.Hour}, r.tmp)
startSlowReceive(t, r, in, "a", 1)
startSlowReceive(t, r, in, "a", 2)
_, err := in.Accept(context.Background(), "a", "t", strings.NewReader("x"))
requireBusy(t, err) // a third body from one user
startSlowReceive(t, r, in, "b", 3)
startSlowReceive(t, r, in, "b", 4)
_, err = in.Accept(context.Background(), "c", "t", strings.NewReader("x"))
requireBusy(t, err) // a fifth body overall
}
+37 -2
View File
@@ -10,6 +10,7 @@ import (
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
"regexp"
"strings" "strings"
"testing" "testing"
) )
@@ -40,10 +41,13 @@ func scratchModule(t *testing.T) string {
return dir return dir
} }
// ansi matches the colour escape sequences task prints when it thinks it has a terminal.
var ansi = regexp.MustCompile("\x1b\\[[0-9;]*m")
// hasSkipNotice reports whether the skip notice was printed as output. Task echoes the command // hasSkipNotice reports whether the skip notice was printed as output. Task echoes the command
// line first, and that line contains the same words, so a substring match would be vacuous. // line first, and that line contains the same words, so a substring match would be vacuous.
func hasSkipNotice(out string) bool { func hasSkipNotice(out string) bool {
for _, line := range strings.Split(out, "\n") { for _, line := range strings.Split(ansi.ReplaceAllString(out, ""), "\n") {
if strings.HasPrefix(line, "golangci-lint not installed") { if strings.HasPrefix(line, "golangci-lint not installed") {
return true return true
} }
@@ -53,12 +57,17 @@ func hasSkipNotice(out string) bool {
// runLint runs `task lint` in dir with the given PATH and returns its combined output and exit code. // runLint runs `task lint` in dir with the given PATH and returns its combined output and exit code.
func runLint(t *testing.T, dir, path string) (string, int) { func runLint(t *testing.T, dir, path string) (string, int) {
t.Helper()
return runTask(t, dir, path, "lint")
}
func runTask(t *testing.T, dir, path string, args ...string) (string, int) {
t.Helper() t.Helper()
taskBin, err := exec.LookPath("task") taskBin, err := exec.LookPath("task")
if err != nil { if err != nil {
t.Skip("task not installed") t.Skip("task not installed")
} }
cmd := exec.Command(taskBin, "lint") cmd := exec.Command(taskBin, args...)
cmd.Dir = dir cmd.Dir = dir
cmd.Env = append(os.Environ(), "PATH="+path) cmd.Env = append(os.Environ(), "PATH="+path)
var out bytes.Buffer var out bytes.Buffer
@@ -87,6 +96,20 @@ func TestTaskLint_ExitsNonZeroWhenGolangciLintReportsAFinding(t *testing.T) {
if hasSkipNotice(out) { if hasSkipNotice(out) {
t.Fatalf("task lint blamed a missing binary for a lint failure; output:\n%s", out) t.Fatalf("task lint blamed a missing binary for a lint failure; output:\n%s", out)
} }
if !strings.Contains(out, "errcheck") {
t.Fatalf("a non-zero exit without the finding means the linter crashed, not that it found something; output:\n%s", out)
}
}
// The lint step only gates anything if `task check` runs it.
func TestTaskCheck_RunsTheLintStep(t *testing.T) {
out, code := runTask(t, scratchModule(t), os.Getenv("PATH"), "--dry", "check")
if code != 0 {
t.Fatalf("task --dry check exited %d; output:\n%s", code, out)
}
if !strings.Contains(out, "task: [lint]") {
t.Fatalf("task check no longer runs the lint step; output:\n%s", out)
}
} }
func TestTaskLint_SkipsWithANoticeWhenGolangciLintIsAbsent(t *testing.T) { func TestTaskLint_SkipsWithANoticeWhenGolangciLintIsAbsent(t *testing.T) {
@@ -106,3 +129,15 @@ func TestTaskLint_SkipsWithANoticeWhenGolangciLintIsAbsent(t *testing.T) {
t.Fatalf("expected the skip notice; output:\n%s", out) t.Fatalf("expected the skip notice; output:\n%s", out)
} }
} }
// CI runs task with colour on, so the notice line starts with an escape sequence; a bare
// prefix match missed it there while passing on a terminal-less local run.
func TestHasSkipNotice_SeesTheNoticeBehindColourCodes(t *testing.T) {
out := "\x1b[32mtask: [lint] ...\n\x1b[0mgolangci-lint not installed — skipping (CI installs it)\n"
if !hasSkipNotice(out) {
t.Fatalf("notice not found in coloured output: %q", out)
}
if hasSkipNotice("\x1b[32mtask: [lint] if command -v golangci-lint ...; echo \"golangci-lint not installed\"\n") {
t.Fatal("the echoed command line must not count as the notice")
}
}