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)         {}