diff --git a/internal/uploads/worker.go b/internal/uploads/worker.go index 6a49646..f6914bd 100644 --- a/internal/uploads/worker.go +++ b/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{}{}: