fix(scheduler): cache transcripts on the scheduled path (ADR-021 regression)
buildUserRunner built the engine without engine.Transcripts = st, so the scheduler — unlike the web "Summarize now" path — never read or wrote the shared transcript cache. Every discovery pass re-fetched transcripts it had already fetched, burning the scarce per-egress-IP caption budget (ADR-014) on redundant work and starving other users' first-time fetches. The transcripts table was empty despite summaries existing. Wire the cache on this path too. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
+37
-2
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user