From 27a88a0a1fb84a8059eda4703ac255f5260877d2 Mon Sep 17 00:00:00 2001 From: Mathias Date: Fri, 2 Oct 2026 08:53:10 +0200 Subject: [PATCH] feat(uploads): the worker reports real phases, beats on its own, and logs each one (ADR-035) Co-Authored-By: Claude Sonnet 5.5 --- internal/uploads/worker.go | 117 ++++++++++++++++++++++++++++++++++--- 1 file changed, 109 insertions(+), 8 deletions(-) diff --git a/internal/uploads/worker.go b/internal/uploads/worker.go index 7b32ab0..334efd3 100644 --- a/internal/uploads/worker.go +++ b/internal/uploads/worker.go @@ -10,6 +10,7 @@ import ( "time" "git.d-ma.be/mathias/tapir/internal/domain" + "git.d-ma.be/mathias/tapir/internal/progress" ) // ErrStopped is returned by Submit once the worker has stopped. @@ -26,13 +27,52 @@ type Transcriber interface { Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) } +// Planner is an optional extra on a Transcriber: it declares its own phases (for +// example one per engine) so the page can show them. A Transcriber that does not +// implement it gets a single "transcribing" phase from the worker. Whatever a Planner +// declares it reports itself, through progress.From(ctx) (ADR-035). +type Planner interface { + Plan() []progress.Planned +} + // 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 + SaveProgress(ctx context.Context, userID, id string, steps []domain.ProgressStep, at time.Time) error + Heartbeat(ctx context.Context, userID, id string) error } +// WorkerOption tunes a Worker. +type WorkerOption func(*Worker) + +// WithHeartbeat sets how often a running job stamps its row alive and how often it logs +// which phase is still running. The defaults are 10 s and 5 min (ADR-035). +func WithHeartbeat(beat, still time.Duration) WorkerOption { + return func(w *Worker) { w.beatEvery, w.stillEvery = beat, still } +} + +// WithLogger replaces slog.Default() as the worker's logger. +func WithLogger(l *slog.Logger) WorkerOption { return func(w *Worker) { w.log = l } } + +const ( + defaultBeatEvery = 10 * time.Second + defaultStillEvery = 5 * time.Minute +) + +// jobSink binds the store to one job so a progress.Tracker can persist through it. +type jobSink struct { + st WorkerStore + userID, id string +} + +func (s jobSink) SaveProgress(ctx context.Context, steps []domain.ProgressStep, at time.Time) error { + return s.st.SaveProgress(ctx, s.userID, s.id, steps, at) +} + +func (s jobSink) Beat(ctx context.Context) error { return s.st.Heartbeat(ctx, s.userID, s.id) } + // 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 @@ -45,6 +85,9 @@ type Worker struct { tmpDir string timeout time.Duration + log *slog.Logger + beatEvery, stillEvery time.Duration + mu sync.Mutex queue []Job stopped bool @@ -53,8 +96,15 @@ type Worker 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)} +func NewWorker(s WorkerStore, c Converter, t Transcriber, tmpDir string, timeout time.Duration, opts ...WorkerOption) *Worker { + w := &Worker{ + store: s, conv: c, tr: t, tmpDir: tmpDir, timeout: timeout, wake: make(chan struct{}, 1), + beatEvery: defaultBeatEvery, stillEvery: defaultStillEvery, + } + for _, o := range opts { + o(w) + } + return w } // Submit queues a job. It never blocks. It satisfies Submitter. @@ -125,9 +175,17 @@ func (w *Worker) stop() { // 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. +// +// Every phase is reported through one progress.Tracker, which saves it to the row for +// the page and logs it, and a heartbeat stamps the row on its own while the job runs +// (ADR-035). Nothing logged or saved here carries transcript text or the file's name. func (w *Worker) Process(ctx context.Context, job Job) { defer func() { _ = os.Remove(job.Path) }() - log := slog.Default().With("upload", job.UploadID) + base := w.log + if base == nil { + base = slog.Default() + } + log := base.With("upload", job.UploadID) start := time.Now() if err := w.store.StartTranscribing(ctx, job.UserID, job.UploadID); err != nil { @@ -136,32 +194,75 @@ func (w *Worker) Process(ctx context.Context, job Job) { return } + plan := []progress.Planned{{Key: "received", Label: "Received"}, {Key: "audio", Label: "Preparing the audio"}} + planner, planned := w.tr.(Planner) + if planned { + plan = append(plan, planner.Plan()...) + } else { + plan = append(plan, progress.Planned{Key: "transcribe", Label: "Transcribing"}) + } + plan = append(plan, progress.Planned{Key: "save", Label: "Saving the transcript"}) + tr := progress.NewTracker(ctx, jobSink{st: w.store, userID: job.UserID, id: job.UploadID}, log, plan) + tr.Start() + tr.Begin("received", 0) + tr.Done("received") + + hbCtx, stopBeat := context.WithCancel(ctx) + hbDone := make(chan struct{}) + go func() { + defer close(hbDone) + tr.Heartbeat(hbCtx, w.beatEvery, w.stillEvery) + }() + defer func() { stopBeat(); <-hbDone }() // no heartbeat outlives its 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() { + if ctx.Err() == nil { + tr.Fail() + } + w.fail(ctx, job, log) + } + wav, err := w.tempWAV() if err != nil { log.Error("upload: temp wav", "err", err) - w.fail(ctx, job, log) + abort() return } defer func() { _ = os.Remove(wav) }() jobCtx, cancel := context.WithTimeout(ctx, w.timeout) defer cancel() + tr.Begin("audio", 0) if err := w.conv.ToWAV(jobCtx, job.Path, wav); err != nil { log.Error("upload: convert", "err", err) - w.fail(ctx, job, log) + abort() return } - t, err := w.tr.Transcribe(jobCtx, wav) + tr.Done("audio") + + if !planned { + tr.Begin("transcribe", 0) + } + t, err := w.tr.Transcribe(progress.With(jobCtx, tr), wav) if err != nil { log.Error("upload: transcribe", "err", err, "elapsed", time.Since(start).Round(time.Second)) - w.fail(ctx, job, log) + abort() return } + if !planned { + tr.Done("transcribe") + } + + tr.Begin("save", 0) 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) + abort() return } + tr.Done("save") + tr.Finish() log.Info("upload transcribed", "endpoint", t.Endpoint, "audio_s", t.AudioSeconds, "segments", len(t.Segments), "elapsed", time.Since(start).Round(time.Second)) }