diff --git a/internal/uploads/worker.go b/internal/uploads/worker.go new file mode 100644 index 0000000..7b32ab0 --- /dev/null +++ b/internal/uploads/worker.go @@ -0,0 +1,191 @@ +package uploads + +import ( + "context" + "errors" + "fmt" + "log/slog" + "os" + "sync" + "time" + + "git.d-ma.be/mathias/tapir/internal/domain" +) + +// ErrStopped is returned by Submit once the worker has stopped. +var ErrStopped = errors.New("uploads: worker stopped") + +// 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 +} + +// Transcriber turns a WAV into a timed transcription. The gateway chain +// (iguana primary, berget fallback) implements it. +type Transcriber interface { + Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) +} + +// WorkerStore is the slice of the private transcript store the worker drives. +type WorkerStore interface { + StartTranscribing(ctx context.Context, userID, id string) error + CompleteUpload(ctx context.Context, userID, id string, t domain.Transcription) error + FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error +} + +// Worker transcribes accepted uploads in-process, ONE AT A TIME (ADR-033). It is +// not a second deployable: it shares the web process's lifetime, which is why it +// inherits ADR-018's single-replica assumption — the queue is process-local. The +// concurrency of one is deliberate: two simultaneous 90-minute jobs on iguana were +// never measured, and a Metal OOM there latches the whisper backend (infra#273). +type Worker struct { + store WorkerStore + conv Converter + tr Transcriber + tmpDir string + timeout time.Duration + + mu sync.Mutex + queue []Job + stopped bool + wake chan struct{} +} + +// NewWorker wires a Worker. timeout bounds each job (conversion + transcription); +// a job that outlives it is failed rather than left spinning. +func NewWorker(s WorkerStore, c Converter, t Transcriber, tmpDir string, timeout time.Duration) *Worker { + return &Worker{store: s, conv: c, tr: t, tmpDir: tmpDir, timeout: timeout, wake: make(chan struct{}, 1)} +} + +// Submit queues a job. It never blocks. It satisfies Submitter. +func (w *Worker) Submit(j Job) error { + w.mu.Lock() + defer w.mu.Unlock() + if w.stopped { + return ErrStopped + } + w.queue = append(w.queue, j) + select { + case w.wake <- struct{}{}: + default: + } + return nil +} + +// Run processes queued jobs until ctx ends, then deletes the audio of anything +// still queued and refuses new jobs. The rows of those jobs stay queued or +// transcribing and are failed as interrupted by the start-up sweep +// (store.FailInterruptedUploads): the audio is not retained, so they cannot resume. +func (w *Worker) Run(ctx context.Context) { + defer w.stop() + for { + job, ok := w.next(ctx) + if !ok { + return + } + w.Process(ctx, job) + } +} + +// next blocks for the next job. Cancellation wins over a non-empty queue: a job +// still queued at shutdown must not be STARTED (it would begin a long conversion +// with an already-cancelled context); stop() deletes its audio instead. +func (w *Worker) next(ctx context.Context) (Job, bool) { + for { + if ctx.Err() != nil { + return Job{}, false + } + w.mu.Lock() + if len(w.queue) > 0 { + j := w.queue[0] + w.queue = w.queue[1:] + w.mu.Unlock() + return j, true + } + w.mu.Unlock() + select { + case <-w.wake: + case <-ctx.Done(): + return Job{}, false + } + } +} + +func (w *Worker) stop() { + w.mu.Lock() + defer w.mu.Unlock() + w.stopped = true + for _, j := range w.queue { + _ = os.Remove(j.Path) + } + w.queue = nil +} + +// Process runs one job: row -> transcribing, convert, transcribe, store. The +// uploaded file and the converted WAV are deleted on every path — the audio is +// never retained (ADR-033). A failure is recorded with a TYPED kind; a shutdown +// mid-job is not a failure and leaves the row for the start-up sweep. +func (w *Worker) Process(ctx context.Context, job Job) { + defer func() { _ = os.Remove(job.Path) }() + log := slog.Default().With("upload", job.UploadID) + start := time.Now() + + if err := w.store.StartTranscribing(ctx, job.UserID, job.UploadID); err != nil { + log.Error("upload: start transcribing", "err", err) + w.fail(ctx, job, log) + return + } + + wav, err := w.tempWAV() + if err != nil { + log.Error("upload: temp wav", "err", err) + w.fail(ctx, job, log) + return + } + defer func() { _ = os.Remove(wav) }() + + jobCtx, cancel := context.WithTimeout(ctx, w.timeout) + defer cancel() + if err := w.conv.ToWAV(jobCtx, job.Path, wav); err != nil { + log.Error("upload: convert", "err", err) + w.fail(ctx, job, log) + return + } + t, err := w.tr.Transcribe(jobCtx, wav) + if err != nil { + log.Error("upload: transcribe", "err", err, "elapsed", time.Since(start).Round(time.Second)) + w.fail(ctx, job, log) + return + } + if err := w.store.CompleteUpload(ctx, job.UserID, job.UploadID, t); err != nil { + log.Error("upload: store transcript", "err", err) + w.fail(ctx, job, log) + return + } + log.Info("upload transcribed", "endpoint", t.Endpoint, "audio_s", t.AudioSeconds, + "segments", len(t.Segments), "elapsed", time.Since(start).Round(time.Second)) +} + +// fail records a transcription failure — unless the PARENT context ended, which +// is a shutdown, not a failure: the row is left for the start-up sweep. +func (w *Worker) fail(ctx context.Context, job Job, log *slog.Logger) { + if ctx.Err() != nil { + return + } + if err := w.store.FailUpload(ctx, job.UserID, job.UploadID, domain.UploadErrTranscriptionFailed); err != nil { + log.Error("upload: record failure", "err", err) + } +} + +func (w *Worker) tempWAV() (string, error) { + f, err := os.CreateTemp(w.tmpDir, "wav-*") + if err != nil { + return "", fmt.Errorf("uploads: temp wav: %w", err) + } + name := f.Name() + if err := f.Close(); err != nil { + _ = os.Remove(name) + return "", fmt.Errorf("uploads: temp wav: %w", err) + } + return name, nil +}