diff --git a/cmd/tapir/scheduler.go b/cmd/tapir/scheduler.go index 8b02c25..cc8b9e5 100644 --- a/cmd/tapir/scheduler.go +++ b/cmd/tapir/scheduler.go @@ -36,6 +36,11 @@ func buildUserRunner(cfg config.Config, st *store.Store, secretStore ports.Secre }, secretStore) engine := usecase.NewEngine(src, buildSummarizer(cfg), st) + // Share the transcript cache (ADR-021) on the scheduler path too — without + // this every scheduled pass re-fetches transcripts it already had, burning the + // scarce per-IP caption budget (ADR-014) and starving other users. The web + // "Summarize now" path already sets this; the scheduler omitting it was a bug. + engine.Transcripts = st return runner.New(src, st, engine, userID, log, runner.WithBackoff(cfg.FetchBackoff), @@ -57,6 +62,7 @@ type userLister interface { // isolation). Returns the stats summed across users. func runDiscoveryPass( ctx context.Context, + pass int, lister userLister, runUser func(context.Context, string) (runner.Stats, error), log *slog.Logger, @@ -67,6 +73,13 @@ func runDiscoveryPass( return runner.Stats{} } + // Rotate who goes first each pass. Caption fetches share one per-egress-IP + // rate budget (ADR-014); whoever runs first each pass spends the pre-throttle + // window, so a FIXED user order permanently starves whoever is last (a new + // pilot user got 0 fetches for 12h while the first-listed user got all of + // them). Rotation gives every user the lead slot in turn. + users = rotateUsers(users, pass) + log.Info("scheduler: starting discovery pass", "users", len(users)) var total runner.Stats for _, u := range users { @@ -116,7 +129,8 @@ func runScheduler( return // disabled } - runDiscoveryPass(ctx, lister, runUser, log) + pass := 0 + runDiscoveryPass(ctx, pass, lister, runUser, log) ticker := time.NewTicker(interval) defer ticker.Stop() @@ -125,11 +139,32 @@ func runScheduler( case <-ctx.Done(): return case <-ticker.C: - runDiscoveryPass(ctx, lister, runUser, log) + pass++ + runDiscoveryPass(ctx, pass, lister, runUser, log) } } } +// rotateUsers left-rotates users by pass positions so a different user leads each +// pass. With n users, user i leads on every pass where pass ≡ i (mod n). A pass +// offset that is negative or exceeds n is normalised. Order within the rotation +// is otherwise preserved, so the set of users run is unchanged — only who is +// first (and thus wins the scarce caption-fetch budget) rotates. +func rotateUsers(users []store.UserIdentity, pass int) []store.UserIdentity { + n := len(users) + if n <= 1 { + return users + } + off := ((pass % n) + n) % n + if off == 0 { + return users + } + out := make([]store.UserIdentity, 0, n) + out = append(out, users[off:]...) + out = append(out, users[:off]...) + return out +} + // sumStats adds two passes' stats field-wise, so runDiscoveryPass can report a // per-tick aggregate across all users. func sumStats(a, b runner.Stats) runner.Stats {