fix(progress): stamp the heartbeat with the application clock the page compares it with (#51)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-03 21:22:17 +02:00
co-authored by Claude Sonnet 5.5
parent 813192e5ba
commit b8af7f32c7
3 changed files with 13 additions and 9 deletions
@@ -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) WHERE id = $1 AND user_id = $2`, b, at)
} }
// Heartbeat stamps the upload as alive now. It does not touch progress_at: alive says // Heartbeat stamps the upload as alive at the given time. It does not touch progress_at:
// the service is still reporting, not that work is moving (ADR-035). // alive says the service is still reporting, not that work is moving (ADR-035). The time
func (s *Store) Heartbeat(ctx context.Context, userID, id string) error { // 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", return s.updateUpload(ctx, userID, id, "heartbeat",
`UPDATE uploaded_transcripts SET alive_at = NOW() `UPDATE uploaded_transcripts SET alive_at = $3
WHERE id = $1 AND user_id = $2`) WHERE id = $1 AND user_id = $2`, at)
} }
// FailUpload moves a queued or transcribing upload to failed with a typed reason. A // FailUpload moves a queued or transcribing upload to failed with a typed reason. A
+2 -2
View File
@@ -37,7 +37,7 @@ type Reporter interface {
// is for people, so a failed write is logged and never stops the job. // is for people, so a failed write is logged and never stops the job.
type Sink interface { type Sink interface {
SaveProgress(ctx context.Context, steps []domain.ProgressStep, at time.Time) error 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 ( const (
@@ -221,7 +221,7 @@ func (t *Tracker) Heartbeat(ctx context.Context, beatEvery, stillEvery time.Dura
return return
case <-beat.C: case <-beat.C:
beatCtx, cancel := context.WithTimeout(ctx, t.saveTimeout) beatCtx, cancel := context.WithTimeout(ctx, t.saveTimeout)
err := t.sink.Beat(beatCtx) err := t.sink.Beat(beatCtx, t.now())
cancel() cancel()
if err != nil && ctx.Err() == nil { if err != nil && ctx.Err() == nil {
t.log.Warn("progress: heartbeat not saved", "err", err) t.log.Warn("progress: heartbeat not saved", "err", err)
+4 -2
View File
@@ -47,7 +47,7 @@ type WorkerStore interface {
CompleteUpload(ctx context.Context, userID, id string, t domain.Transcription) error CompleteUpload(ctx context.Context, userID, id string, t domain.Transcription) error
FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) 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 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. // 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) 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 // 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 // not a second deployable: it shares the web process's lifetime, which is why it