diff --git a/internal/uploads/worker_queue_test.go b/internal/uploads/worker_queue_test.go new file mode 100644 index 0000000..259693d --- /dev/null +++ b/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") +}