feat(uploads): the worker measures the converted audio against the limit and the claim (review F2)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-02 16:45:25 +02:00
co-authored by Claude Sonnet 5.5
parent 132f46b047
commit 16cd246241
+52 -9
View File
@@ -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
}