From b8af7f32c79f0b8f62a2f19029fd5a34d13171a0 Mon Sep 17 00:00:00 2001 From: Mathias Date: Sat, 3 Oct 2026 21:04:42 +0200 Subject: [PATCH] fix(progress): stamp the heartbeat with the application clock the page compares it with (#51) Co-Authored-By: Claude Sonnet 5.5 --- internal/adapters/store/uploaded_transcripts.go | 12 +++++++----- internal/progress/tracker.go | 4 ++-- internal/uploads/worker.go | 6 ++++-- 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/internal/adapters/store/uploaded_transcripts.go b/internal/adapters/store/uploaded_transcripts.go index a77d160..c62fb5a 100644 --- a/internal/adapters/store/uploaded_transcripts.go +++ b/internal/adapters/store/uploaded_transcripts.go @@ -80,12 +80,14 @@ func (s *Store) SaveProgress(ctx context.Context, userID, id string, steps []dom WHERE id = $1 AND user_id = $2`, b, at) } -// Heartbeat stamps the upload as alive now. It does not touch progress_at: alive says -// the service is still reporting, not that work is moving (ADR-035). -func (s *Store) Heartbeat(ctx context.Context, userID, id string) error { +// Heartbeat stamps the upload as alive at the given time. It does not touch progress_at: +// alive says the service is still reporting, not that work is moving (ADR-035). The time +// comes from the application clock, the one progress_at and the page's health check use, +// so a database clock that disagrees cannot make a healthy job look lost (#51). +func (s *Store) Heartbeat(ctx context.Context, userID, id string, at time.Time) error { return s.updateUpload(ctx, userID, id, "heartbeat", - `UPDATE uploaded_transcripts SET alive_at = NOW() - WHERE id = $1 AND user_id = $2`) + `UPDATE uploaded_transcripts SET alive_at = $3 + WHERE id = $1 AND user_id = $2`, at) } // FailUpload moves a queued or transcribing upload to failed with a typed reason. A diff --git a/internal/progress/tracker.go b/internal/progress/tracker.go index d8878b4..c365d58 100644 --- a/internal/progress/tracker.go +++ b/internal/progress/tracker.go @@ -37,7 +37,7 @@ type Reporter interface { // is for people, so a failed write is logged and never stops the job. type Sink interface { SaveProgress(ctx context.Context, steps []domain.ProgressStep, at time.Time) error - Beat(ctx context.Context) error + Beat(ctx context.Context, at time.Time) error } const ( @@ -221,7 +221,7 @@ func (t *Tracker) Heartbeat(ctx context.Context, beatEvery, stillEvery time.Dura return case <-beat.C: beatCtx, cancel := context.WithTimeout(ctx, t.saveTimeout) - err := t.sink.Beat(beatCtx) + err := t.sink.Beat(beatCtx, t.now()) cancel() if err != nil && ctx.Err() == nil { t.log.Warn("progress: heartbeat not saved", "err", err) diff --git a/internal/uploads/worker.go b/internal/uploads/worker.go index 9fa2958..a29fecf 100644 --- a/internal/uploads/worker.go +++ b/internal/uploads/worker.go @@ -47,7 +47,7 @@ type WorkerStore interface { 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 + Heartbeat(ctx context.Context, userID, id string, at time.Time) error } // WorkerOption tunes a Worker. @@ -91,7 +91,9 @@ func (s jobSink) SaveProgress(ctx context.Context, steps []domain.ProgressStep, 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) } +func (s jobSink) Beat(ctx context.Context, at time.Time) error { + return s.st.Heartbeat(ctx, s.userID, s.id, at) +} // 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