internal/uploads/worker_queue_test.go

@@ -0,0 +1,80 @@
+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")
+}