Files
tapir/internal/uploads/worker_queue_test.go
T
mathias 4a2067063d
CI / Deploy via GitOps (push) Successful in 7s
CI / Lint / Test / Vet (push) Successful in 33s
CI / Build & Import (push) Successful in 23s
feat: upload audio, get a private Swedish transcript, discuss it (#33, ADR-033) (#39)
Upload a recording (audio, or video with the camera off), get a private Swedish transcript, and chat about it.

- ADR-033: RLS-scoped private transcript store; ffprobe/ffmpeg intake; gateway transcription (iguana kb-whisper, berget fallback); in-process single-replica worker; chat over the transcript (local models only).
- ADR-034: optional two-engine mode (pianissimo + kb-whisper) reconciled by a LOCAL model; off unless TAPIR_RECONCILE_BASE_MODEL is set.
- ADR-035: real progress (phases, running/quiet/lost), phase logging, every judged spot stored, merged segments.
- Hardening from four adversarial reviews: errors logged and stored as kinds never text; per-engine deadlines; converted-length check; bounded queue; client hang-ups are not errors; strict local-alias check.
- Dockerfile: static ffmpeg/ffprobe 7.0.2 pinned by sha256 (#34). Built and inspected by hand; CI does not build the image on a PR, so its first CI build is this merge.

Uploads stay OFF until TAPIR_UPLOADS_ENABLED is set in the deployment.

Closes #33, #34, #42, #43, #44.

Co-authored-by: Mathias <mthbqv@gmail.com>
2026-10-03 14:13:37 +00:00

109 lines
3.6 KiB
Go

package uploads_test
import (
"context"
"errors"
"strconv"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"git.d-ma.be/mathias/tapir/internal/uploads"
)
// The queue was append-only: 100,000 submits with the worker idle all succeeded, and each
// queued job holds up to the upload cap on disk until its turn (adversarial review F6).
func queueWorker(t *testing.T, total, perUser int) *uploads.Worker {
t.Helper()
r := newWRig(t, time.Minute)
return uploads.NewWorker(r.st, r.conv, r.tr, r.tmp, time.Minute, uploads.WithQueueLimits(total, perUser))
}
func submit(w *uploads.Worker, user string, n int) error {
return w.Submit(uploads.Job{UserID: user, UploadID: user + "-" + strconv.Itoa(n), Path: "/nonexistent"})
}
func TestWorkerSubmit_RefusesBeyondTheTotalQueueCap(t *testing.T) {
w := queueWorker(t, 3, 10)
for i := 0; i < 3; i++ {
require.NoError(t, submit(w, "u"+strconv.Itoa(i), i))
}
err := submit(w, "u9", 9)
assert.ErrorIs(t, err, uploads.ErrQueueFull)
}
func TestWorkerSubmit_OneUserCannotFillTheQueueForEveryoneElse(t *testing.T) {
w := queueWorker(t, 100, 2)
require.NoError(t, submit(w, "alice", 1))
require.NoError(t, submit(w, "alice", 2))
errAlice := submit(w, "alice", 3)
errBob := submit(w, "bob", 1)
assert.True(t, errors.Is(errAlice, uploads.ErrQueueFull), "a third waiting upload from one user is refused")
assert.NoError(t, errBob, "another user is not affected by alice's queue")
}
func TestWorkerSubmit_ThereIsADefaultCapSoANeglectedWorkerCannotBeFilledForever(t *testing.T) {
r := newWRig(t, time.Minute)
w := uploads.NewWorker(r.st, r.conv, r.tr, r.tmp, time.Minute)
refused := 0
for i := 0; i < 1000; i++ {
if err := submit(w, "u"+strconv.Itoa(i), i); errors.Is(err, uploads.ErrQueueFull) {
refused++
}
}
assert.Positive(t, refused, "with the defaults, 1,000 submits to an idle worker must not all be accepted")
}
func TestWorkerSubmit_AJobThatHasLeftTheQueueFreesItsPlace(t *testing.T) {
r := newWRig(t, time.Minute)
w := uploads.NewWorker(r.st, r.conv, r.tr, r.tmp, time.Minute, uploads.WithQueueLimits(1, 1))
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { defer close(done); w.Run(ctx) }()
defer func() { cancel(); <-done }()
job := r.spool(t, "first")
require.NoError(t, w.Submit(job))
require.Eventually(t, func() bool { c, _ := r.st.snapshot(); _, ok := c["first"]; return ok },
5*time.Second, 5*time.Millisecond, "the first job never completed")
assert.NoError(t, w.Submit(r.spool(t, "second")), "the cap counts jobs waiting, not jobs ever submitted")
}
func TestWorkerHasRoom_SaysWhetherASubmitWouldBeAccepted(t *testing.T) {
w := queueWorker(t, 3, 2)
assert.True(t, w.HasRoom("alice"))
require.NoError(t, submit(w, "alice", 1))
require.NoError(t, submit(w, "alice", 2))
assert.False(t, w.HasRoom("alice"), "alice is at her share")
assert.True(t, w.HasRoom("bob"), "bob is not")
require.NoError(t, submit(w, "bob", 1))
assert.False(t, w.HasRoom("carol"), "the queue as a whole is at its cap")
}
func TestWorkerHasRoom_AgreesWithSubmitAtEveryStepAndIsFalseOnceStopped(t *testing.T) {
w := queueWorker(t, 2, 1)
for i, user := range []string{"a", "a", "b", "c"} {
room := w.HasRoom(user)
err := submit(w, user, i)
assert.Equal(t, room, err == nil, "step %d (%s): HasRoom=%v but Submit returned %v", i, user, room, err)
}
r := newWRig(t, time.Minute)
stopped := uploads.NewWorker(r.st, r.conv, r.tr, r.tmp, time.Minute)
ctx, cancel := context.WithCancel(context.Background())
cancel()
stopped.Run(ctx)
assert.False(t, stopped.HasRoom("a"), "a worker that has stopped has no room")
}