diff --git a/internal/progress/tracker.go b/internal/progress/tracker.go new file mode 100644 index 0000000..0c6e0d0 --- /dev/null +++ b/internal/progress/tracker.go @@ -0,0 +1,246 @@ +// Package progress tracks the phases of one upload job (ADR-035). A Tracker is the +// only writer of a job's progress: every transition is saved to the job's row and +// logged by the same call, so what the page shows and what the log says cannot +// disagree. The engines and the reconciler report through a Reporter carried in the +// context, so the Transcriber interface stays what it was. +package progress + +import ( + "context" + "log/slog" + "sync" + "time" + + "git.d-ma.be/mathias/tapir/internal/domain" +) + +// Planned is one phase of a job, declared up front so the page can show what is coming. +type Planned struct { + Key, Label string +} + +// Reporter is what a long-running component reports to. Begin starts a phase (total +// is the number of countable units, 0 when there is no honest count), Advance says how +// many are done, Done closes the phase. +type Reporter interface { + Begin(key string, total int) + Advance(key string, done int) + Done(key string) +} + +// Sink is where a job's progress is persisted. Both calls are best effort: progress +// 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 +} + +const ( + defaultMinSave = 2 * time.Second + logAdvanceEach = time.Minute +) + +// Option tunes a Tracker. +type Option func(*Tracker) + +// WithClock fixes the clock, for tests. +func WithClock(now func() time.Time) Option { return func(t *Tracker) { t.now = now } } + +// WithMinSave sets the least time between two saves caused by Advance alone. Begin, +// Done, Fail and Finish always save. +func WithMinSave(d time.Duration) Option { return func(t *Tracker) { t.minSave = d } } + +// Tracker implements Reporter for one job. +type Tracker struct { + ctx context.Context + sink Sink + log *slog.Logger + now func() time.Time + minSave time.Duration + + mu sync.Mutex + steps []domain.ProgressStep + lastAdvanceSave time.Time + lastAdvanceLog time.Time +} + +// 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} + for _, o := range opts { + o(t) + } + for _, p := range plan { + t.steps = append(t.steps, domain.ProgressStep{Key: p.Key, Label: p.Label, State: domain.StepPending}) + } + return t +} + +// Start saves the plan, every phase pending. +func (t *Tracker) Start() { + t.mu.Lock() + defer t.mu.Unlock() + t.log.Info("job started", "steps", len(t.steps)) + t.save() +} + +// Begin implements Reporter. +func (t *Tracker) Begin(key string, total int) { + t.mu.Lock() + defer t.mu.Unlock() + s := t.find(key) + if s == nil { + return + } + s.State, s.StartedAt, s.FinishedAt, s.Done, s.Total = domain.StepRunning, t.now(), time.Time{}, 0, total + t.log.Info("step started", "step", key, "total", total) + t.save() +} + +// Advance implements Reporter. The count is always kept; it is saved at most once per +// minimum interval and logged at most once a minute, so a counted phase does not turn +// into a write and a log line per unit. +func (t *Tracker) Advance(key string, done int) { + t.mu.Lock() + defer t.mu.Unlock() + s := t.find(key) + if s == nil { + return + } + s.Done = done + now := t.now() + if t.lastAdvanceLog.IsZero() || now.Sub(t.lastAdvanceLog) >= logAdvanceEach { + t.lastAdvanceLog = now + t.log.Info("step progress", "step", key, "done", done, "total", s.Total) + } + if t.lastAdvanceSave.IsZero() || now.Sub(t.lastAdvanceSave) >= t.minSave { + t.lastAdvanceSave = now + t.save() + } +} + +// Done implements Reporter. +func (t *Tracker) Done(key string) { + t.mu.Lock() + defer t.mu.Unlock() + s := t.find(key) + if s == nil { + return + } + t.close(s, domain.StepDone) + t.log.Info("step finished", "step", key, "elapsed", s.FinishedAt.Sub(s.StartedAt)) + t.save() +} + +// Fail marks the phase that was running as failed. What never started stays pending, +// which is what shows the user where the job stopped. +func (t *Tracker) Fail() { + t.mu.Lock() + defer t.mu.Unlock() + for i := range t.steps { + if t.steps[i].State == domain.StepRunning { + t.close(&t.steps[i], domain.StepFailed) + t.log.Warn("step failed", "step", t.steps[i].Key, "elapsed", t.steps[i].FinishedAt.Sub(t.steps[i].StartedAt)) + } + } + t.save() +} + +// Finish closes the job: a phase still running is done, a phase that never ran is +// skipped. +func (t *Tracker) Finish() { + t.mu.Lock() + defer t.mu.Unlock() + for i := range t.steps { + switch t.steps[i].State { + case domain.StepRunning: + t.close(&t.steps[i], domain.StepDone) + case domain.StepPending: + t.steps[i].State = domain.StepSkipped + } + } + t.log.Info("job finished") + t.save() +} + +// Heartbeat stamps the job's row every beatEvery and logs which phase is still running +// every stillEvery, until ctx ends. It runs on its own, independent of any engine call: +// a heartbeat that stopped means the process is gone, not that a request is slow. +func (t *Tracker) Heartbeat(ctx context.Context, beatEvery, stillEvery time.Duration) { + beat, still := time.NewTicker(beatEvery), time.NewTicker(stillEvery) + defer beat.Stop() + defer still.Stop() + for { + select { + case <-ctx.Done(): + return + case <-beat.C: + if err := t.sink.Beat(ctx); err != nil && ctx.Err() == nil { + t.log.Warn("progress: heartbeat not saved", "err", err) + } + case <-still.C: + t.logStill() + } + } +} + +func (t *Tracker) logStill() { + t.mu.Lock() + defer t.mu.Unlock() + for _, s := range t.steps { + if s.State == domain.StepRunning { + t.log.Info("still running", "step", s.Key, "elapsed", t.now().Sub(s.StartedAt), "done", s.Done, "total", s.Total) + return + } + } +} + +// close ends a phase. Callers hold t.mu. +func (t *Tracker) close(s *domain.ProgressStep, state domain.StepState) { + s.State, s.FinishedAt = state, t.now() + if s.StartedAt.IsZero() { + s.StartedAt = s.FinishedAt + } +} + +// find returns the phase with this key, or nil. Callers hold t.mu. +func (t *Tracker) find(key string) *domain.ProgressStep { + for i := range t.steps { + if t.steps[i].Key == key { + return &t.steps[i] + } + } + return nil +} + +// save persists a copy of the phases. Callers hold t.mu. A failed write is logged and +// 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 { + t.log.Warn("progress: not saved", "err", err) + } +} + +type ctxKey struct{} + +// With returns a context that carries r. +func With(ctx context.Context, r Reporter) context.Context { + return context.WithValue(ctx, ctxKey{}, r) +} + +// From returns the Reporter carried by ctx, or one that does nothing, so a component +// can always report without checking whether anyone is listening. +func From(ctx context.Context) Reporter { + if r, ok := ctx.Value(ctxKey{}).(Reporter); ok { + return r + } + return noop{} +} + +type noop struct{} + +func (noop) Begin(string, int) {} +func (noop) Advance(string, int) {} +func (noop) Done(string) {}