internal/uploads/worker.go

@@ -17,6 +17,11 @@ import (
 // ErrStopped is returned by Submit once the worker has stopped.
 var ErrStopped = errors.New("uploads: worker stopped")
 
+// ErrQueueFull is returned by Submit when the queue, or one user's share of it, is at
+// its cap. Every queued job holds up to the upload cap on disk until its turn, so an
+// unbounded queue is a disk-filling lever (ADR-035 review).
+var ErrQueueFull = errors.New("uploads: queue full")
+
 // Converter turns the uploaded file into the 16 kHz mono PCM WAV whisper wants.
 type Converter interface {
 	ToWAV(ctx context.Context, src, dst string) error
@@ -54,6 +59,12 @@ func WithHeartbeat(beat, still time.Duration) WorkerOption {
 	return func(w *Worker) { w.beatEvery, w.stillEvery = beat, still }
 }
 
+// WithQueueLimits bounds how many jobs may WAIT: in total, and per user. A job being
+// processed is not waiting. The defaults are 20 and 3 (one instance, one job at a time).
+func WithQueueLimits(total, perUser int) WorkerOption {
+	return func(w *Worker) { w.maxQueued, w.maxPerUser = total, perUser }
+}
+
 // WithMaxDuration makes the worker measure the converted audio and fail a job whose real
 // length exceeds the limit, or falls far short of what the file claimed. Zero (the
 // default) measures nothing. The limit is first checked against the file's header, which
@@ -64,6 +75,8 @@ func WithMaxDuration(d time.Duration) WorkerOption { return func(w *Worker) { w.
 func WithLogger(l *slog.Logger) WorkerOption { return func(w *Worker) { w.log = l } }
 
 const (
+	defaultMaxQueued  = 20
+	defaultMaxPerUser = 3
 	defaultBeatEvery  = 10 * time.Second
 	defaultStillEvery = 5 * time.Minute
 )
@@ -95,6 +108,7 @@ type Worker struct {
 	log                   *slog.Logger
 	beatEvery, stillEvery time.Duration
 	maxDuration           time.Duration
+	maxQueued, maxPerUser int
 
 	mu      sync.Mutex
 	queue   []Job
@@ -108,6 +122,7 @@ func NewWorker(s WorkerStore, c Converter, t Transcriber, tmpDir string, timeout
 	w := &Worker{
 		store: s, conv: c, tr: t, tmpDir: tmpDir, timeout: timeout, wake: make(chan struct{}, 1),
 		beatEvery: defaultBeatEvery, stillEvery: defaultStillEvery,
+		maxQueued: defaultMaxQueued, maxPerUser: defaultMaxPerUser,
 	}
 	for _, o := range opts {
 		o(w)
@@ -122,6 +137,18 @@ func (w *Worker) Submit(j Job) error {
 	if w.stopped {
 		return ErrStopped
 	}
+	if len(w.queue) >= w.maxQueued {
+		return ErrQueueFull
+	}
+	mine := 0
+	for _, q := range w.queue {
+		if q.UserID == j.UserID {
+			mine++
+		}
+	}
+	if mine >= w.maxPerUser {
+		return ErrQueueFull
+	}
 	w.queue = append(w.queue, j)
 	select {
 	case w.wake <- struct{}{}: