From 5644dc3fc87005e1ee72fc6096befcfe648640f0 Mon Sep 17 00:00:00 2001 From: Mathias Date: Sat, 3 Oct 2026 21:00:55 +0200 Subject: [PATCH] fix(uploads): record a job interrupted by shutdown at once, on its own short context (#50) Co-Authored-By: Claude Sonnet 5.5 --- internal/uploads/worker.go | 25 +++++++++++++++++-------- 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/internal/uploads/worker.go b/internal/uploads/worker.go index 8d4b738..9fa2958 100644 --- a/internal/uploads/worker.go +++ b/internal/uploads/worker.go @@ -176,9 +176,10 @@ func (w *Worker) Submit(j Job) error { } // 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. +// still queued and refuses new jobs. The job running at shutdown is recorded interrupted +// by Process; the rows of jobs still queued stay queued 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 { @@ -227,7 +228,7 @@ func (w *Worker) stop() { // 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. +// mid-job is recorded as interrupted at once (see fail). // // 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 @@ -269,7 +270,7 @@ func (w *Worker) Process(ctx context.Context, job Job) { 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. + // a failure of the step: it stays as it was, and fail records the row as interrupted. abort := func(kind domain.UploadErrorKind) { if ctx.Err() == nil { tr.FailRunning() @@ -332,11 +333,19 @@ func (w *Worker) Process(ctx context.Context, job Job) { "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. +// interruptedWriteTimeout bounds the one write a dying process makes to record its job. +const interruptedWriteTimeout = 3 * time.Second + +// fail records a transcription failure. When the PARENT context ended the process is +// shutting down: the job is recorded interrupted at once, on a short context of its own, +// because the process may exit before any other pod's start-up sweep would do it (#50). +// The sweep remains the backstop for a process that is killed outright. func (w *Worker) fail(ctx context.Context, job Job, log *slog.Logger, kind domain.UploadErrorKind) { if ctx.Err() != nil { - return + kind = domain.UploadErrInterrupted + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(context.WithoutCancel(ctx), interruptedWriteTimeout) + defer cancel() } if err := w.store.FailUpload(ctx, job.UserID, job.UploadID, kind); err != nil { log.Error("upload: record failure", "kind", errkind.Of(err))