diff --git a/internal/uploads/worker.go b/internal/uploads/worker.go index eb6d78c..6a49646 100644 --- a/internal/uploads/worker.go +++ b/internal/uploads/worker.go @@ -54,6 +54,12 @@ func WithHeartbeat(beat, still time.Duration) WorkerOption { return func(w *Worker) { w.beatEvery, w.stillEvery = beat, still } } +// 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 +// can lie (ADR-035 review). +func WithMaxDuration(d time.Duration) WorkerOption { return func(w *Worker) { w.maxDuration = d } } + // WithLogger replaces slog.Default() as the worker's logger. func WithLogger(l *slog.Logger) WorkerOption { return func(w *Worker) { w.log = l } } @@ -88,6 +94,7 @@ type Worker struct { log *slog.Logger beatEvery, stillEvery time.Duration + maxDuration time.Duration mu sync.Mutex queue []Job @@ -191,7 +198,7 @@ func (w *Worker) Process(ctx context.Context, job Job) { if err := w.store.StartTranscribing(ctx, job.UserID, job.UploadID); err != nil { log.Error("upload: start transcribing", "err", err) - w.fail(ctx, job, log) + w.fail(ctx, job, log, domain.UploadErrTranscriptionFailed) return } @@ -218,17 +225,17 @@ func (w *Worker) Process(ctx context.Context, job Job) { // abort records where the job stopped. A shutdown (the parent context ended) is not // a failure: the row is left for the start-up sweep, so the step stays as it was. - abort := func() { + abort := func(kind domain.UploadErrorKind) { if ctx.Err() == nil { tr.FailRunning() } - w.fail(ctx, job, log) + w.fail(ctx, job, log, kind) } wav, err := w.tempWAV() if err != nil { log.Error("upload: temp wav", "err", err) - abort() + abort(domain.UploadErrTranscriptionFailed) return } defer func() { _ = os.Remove(wav) }() @@ -238,7 +245,12 @@ func (w *Worker) Process(ctx context.Context, job Job) { tr.Begin("audio", 0) if err := w.conv.ToWAV(jobCtx, job.Path, wav); err != nil { log.Error("upload: convert", "kind", errkind.Of(err)) - abort() + abort(domain.UploadErrTranscriptionFailed) + return + } + if kind, bad := w.audioLengthProblem(wav, job.Duration); bad { + abort(kind) + log.Error("upload: converted audio length rejected", "kind", string(kind)) return } tr.Done("audio") @@ -249,7 +261,7 @@ func (w *Worker) Process(ctx context.Context, job Job) { t, err := w.tr.Transcribe(progress.With(jobCtx, tr), wav) if err != nil { log.Error("upload: transcribe", "kind", errkind.Of(err), "elapsed", time.Since(start).Round(time.Second)) - abort() + abort(domain.UploadErrTranscriptionFailed) return } if !planned { @@ -259,7 +271,7 @@ func (w *Worker) Process(ctx context.Context, job Job) { tr.Begin("save", 0) if err := w.store.CompleteUpload(ctx, job.UserID, job.UploadID, t); err != nil { log.Error("upload: store transcript", "err", err) - abort() + abort(domain.UploadErrTranscriptionFailed) return } tr.Done("save") @@ -270,11 +282,11 @@ func (w *Worker) Process(ctx context.Context, job Job) { // 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) { +func (w *Worker) fail(ctx context.Context, job Job, log *slog.Logger, kind domain.UploadErrorKind) { if ctx.Err() != nil { return } - if err := w.store.FailUpload(ctx, job.UserID, job.UploadID, domain.UploadErrTranscriptionFailed); err != nil { + if err := w.store.FailUpload(ctx, job.UserID, job.UploadID, kind); err != nil { log.Error("upload: record failure", "err", err) } } @@ -291,3 +303,34 @@ func (w *Worker) tempWAV() (string, error) { } return name, nil } + +const ( + // tooLongSlack is how far over the limit the converted audio may be before the claim + // in the file's header is called a lie: ffmpeg rounds. + tooLongSlack = 30 * time.Second + // A file counts as truncated when it converts to under 80% of the length it claimed + // AND more than two minutes short. Probe estimates for variable-bitrate files are + // imprecise; refusing real recordings would be worse than the problem. + incompleteBelow = 0.8 + incompleteShortfall = 2 * time.Minute +) + +// audioLengthProblem measures the converted WAV (16 kHz mono 16-bit PCM is 32,000 bytes a +// second after a 44-byte header) against the limit and against what the file claimed. +func (w *Worker) audioLengthProblem(wav string, claimed time.Duration) (domain.UploadErrorKind, bool) { + if w.maxDuration <= 0 { + return "", false + } + info, err := os.Stat(wav) + if err != nil { + return "", false + } + got := time.Duration(float64(max(info.Size()-44, 0)) / 32000 * float64(time.Second)) + switch { + case got > w.maxDuration+tooLongSlack: + return domain.UploadErrTooLong, true + case float64(got) < incompleteBelow*float64(claimed) && claimed-got > incompleteShortfall: + return domain.UploadErrIncomplete, true + } + return "", false +}