diff --git a/internal/progress/tracker.go b/internal/progress/tracker.go index 71c3d2f..d8878b4 100644 --- a/internal/progress/tracker.go +++ b/internal/progress/tracker.go @@ -41,8 +41,9 @@ type Sink interface { } const ( - defaultMinSave = 2 * time.Second - logAdvanceEach = time.Minute + defaultMinSave = 2 * time.Second + defaultSaveTimeout = 5 * time.Second + logAdvanceEach = time.Minute ) // Option tunes a Tracker. @@ -55,6 +56,11 @@ func WithClock(now func() time.Time) Option { return func(t *Tracker) { t.now = // Done, Fail and Finish always save. func WithMinSave(d time.Duration) Option { return func(t *Tracker) { t.minSave = d } } +// WithSaveTimeout bounds one save or heartbeat write. Progress is cosmetic, so a database +// that stops answering costs a dropped write, never a stalled job: saves run under the +// tracker's lock, which Begin and Advance (called by the engines) need (#51). +func WithSaveTimeout(d time.Duration) Option { return func(t *Tracker) { t.saveTimeout = d } } + // Tracker implements Reporter for one job. type Tracker struct { ctx context.Context @@ -63,6 +69,8 @@ type Tracker struct { now func() time.Time minSave time.Duration + saveTimeout time.Duration + mu sync.Mutex steps []domain.ProgressStep lastAdvanceSave time.Time @@ -72,7 +80,7 @@ type Tracker struct { // NewTracker plans a job's phases. ctx is used for the saves, so it should outlive the // job's own timeout: a job that timed out still gets its failure recorded. func NewTracker(ctx context.Context, sink Sink, log *slog.Logger, plan []Planned, opts ...Option) *Tracker { - t := &Tracker{ctx: ctx, sink: sink, log: log, now: time.Now, minSave: defaultMinSave} + t := &Tracker{ctx: ctx, sink: sink, log: log, now: time.Now, minSave: defaultMinSave, saveTimeout: defaultSaveTimeout} for _, o := range opts { o(t) } @@ -212,7 +220,10 @@ func (t *Tracker) Heartbeat(ctx context.Context, beatEvery, stillEvery time.Dura case <-ctx.Done(): return case <-beat.C: - if err := t.sink.Beat(ctx); err != nil && ctx.Err() == nil { + beatCtx, cancel := context.WithTimeout(ctx, t.saveTimeout) + err := t.sink.Beat(beatCtx) + cancel() + if err != nil && ctx.Err() == nil { t.log.Warn("progress: heartbeat not saved", "err", err) } case <-still.C: @@ -254,7 +265,9 @@ func (t *Tracker) find(key string) *domain.ProgressStep { // dropped: progress is cosmetic and must never fail the job it describes. func (t *Tracker) save() { snapshot := append([]domain.ProgressStep(nil), t.steps...) - if err := t.sink.SaveProgress(t.ctx, snapshot, t.now()); err != nil && t.ctx.Err() == nil { + ctx, cancel := context.WithTimeout(t.ctx, t.saveTimeout) + defer cancel() + if err := t.sink.SaveProgress(ctx, snapshot, t.now()); err != nil && t.ctx.Err() == nil { t.log.Warn("progress: not saved", "err", err) } }