From 40a614c8d4f3454802a53d039f95834b615d9b71 Mon Sep 17 00:00:00 2001 From: Mathias Date: Wed, 3 Jun 2026 22:56:07 +0200 Subject: [PATCH] =?UTF-8?q?feat(runner):=20429=20backoff=20=E2=80=94=20ski?= =?UTF-8?q?p=20still-throttled=20videos,=20persist=20status?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After ProcessNewVideo the runner records transcript_status per outcome: rate_limited (stamps the backoff clock), none, or fetched. Before fetching, a video inside the TAPIR_FETCH_BACKOFF window is skipped (SkippedRateLimited) so a just-429'd caption endpoint is not re-hit; once the window expires it retries. Backoff/clock injected via variadic Options (WithBackoff, WithClock) so existing New call sites and the fake-driven loop tests stay valid. Backoff 0 = always retry. Co-Authored-By: Claude Opus 4.8 (1M context) --- cmd/tapir/main.go | 4 +- internal/runner/runner.go | 95 +++++++++++++++++++++++++++++----- internal/runner/runner_test.go | 85 ++++++++++++++++++++++++++++-- 3 files changed, 163 insertions(+), 21 deletions(-) diff --git a/cmd/tapir/main.go b/cmd/tapir/main.go index f8c2384..f368016 100644 --- a/cmd/tapir/main.go +++ b/cmd/tapir/main.go @@ -125,10 +125,10 @@ func cmdRun(ctx context.Context, log *slog.Logger) error { if engine == nil { return fmt.Errorf("run: incomplete summarization config (gateway, youtube credentials, secrets file)") } - r := runner.New(engine.Source, st, engine, cfg.UserID, log) + r := runner.New(engine.Source, st, engine, cfg.UserID, log, runner.WithBackoff(cfg.FetchBackoff)) log.Info("starting run", "user", cfg.UserID, "model", cfg.SummarizerModel, - "gateway", cfg.GatewayURL, "poll_interval", cfg.PollInterval) + "gateway", cfg.GatewayURL, "poll_interval", cfg.PollInterval, "fetch_backoff", cfg.FetchBackoff) return r.Loop(ctx, cfg.PollInterval) } diff --git a/internal/runner/runner.go b/internal/runner/runner.go index c0e3dfc..b85b504 100644 --- a/internal/runner/runner.go +++ b/internal/runner/runner.go @@ -32,6 +32,12 @@ type VideoStore interface { GetAutoSummarize(ctx context.Context, userID string) (bool, error) RequestedVideoIDs(ctx context.Context, userID string) (map[string]bool, error) ClearSummarizeRequested(ctx context.Context, userID, videoID string) error + // RateLimitedVideoIDs maps the user's still-throttled videos to when they were + // rate-limited, so the loop can back off without re-hitting the caption endpoint. + RateLimitedVideoIDs(ctx context.Context, userID string) (map[string]time.Time, error) + // SetTranscriptStatus records the outcome of a transcript attempt: "none", + // "rate_limited" (stamps the backoff clock), or "fetched". + SetTranscriptStatus(ctx context.Context, userID, videoID, status string) error } // Processor runs the core use case for a single video. *usecase.Engine @@ -43,29 +49,51 @@ type Processor interface { // Runner walks a user's subscriptions, persists each candidate video, skips the // ones already summarized (durably), and processes the rest through the engine. type Runner struct { - src ports.VideoSource - store VideoStore - engine Processor - userID string - log *slog.Logger + src ports.VideoSource + store VideoStore + engine Processor + userID string + log *slog.Logger + backoff time.Duration // rate-limit retry window; 0 = always retry + now func() time.Time // injectable clock (tests); defaults to time.Now } +// Option configures a Runner at construction. Variadic so existing call sites +// stay valid as new knobs (backoff, clock) are added. +type Option func(*Runner) + +// WithBackoff sets the rate-limit retry window. A video that returned HTTP 429 is +// skipped (no caption fetch) until this much time has passed; 0 = always retry. +func WithBackoff(d time.Duration) Option { return func(r *Runner) { r.backoff = d } } + +// WithClock overrides the clock used for backoff comparisons. Tests inject a +// fixed time; production leaves the time.Now default. +func WithClock(now func() time.Time) Option { return func(r *Runner) { r.now = now } } + // New builds a Runner. A nil logger falls back to slog.Default. -func New(src ports.VideoSource, store VideoStore, engine Processor, userID string, log *slog.Logger) *Runner { +func New(src ports.VideoSource, store VideoStore, engine Processor, userID string, log *slog.Logger, opts ...Option) *Runner { if log == nil { log = slog.Default() } - return &Runner{src: src, store: store, engine: engine, userID: userID, log: log} + r := &Runner{src: src, store: store, engine: engine, userID: userID, log: log, now: time.Now} + for _, opt := range opts { + opt(r) + } + if r.now == nil { + r.now = time.Now + } + return r } // Stats summarizes one RunOnce pass. type Stats struct { - Candidates int - Summarized int - SkippedSeen int - SkippedNoText int - SkippedManual int // discovered but not queued, in manual mode - Errors int + Candidates int + Summarized int + SkippedSeen int + SkippedNoText int + SkippedManual int // discovered but not queued, in manual mode + SkippedRateLimited int // 429'd previously and still inside the backoff window + Errors int } // RunOnce performs a single pass over the user's subscriptions. Per-item errors @@ -104,6 +132,18 @@ func (r *Runner) RunOnce(ctx context.Context) (Stats, error) { } } + // Rate-limit backoff: videos that 429'd on a prior pass, mapped to when. Inside + // the backoff window they are skipped before any caption fetch, so a throttled + // IP is not hammered. Loaded once per pass (like seen/requested). Disabled when + // backoff <= 0 ("always retry"). + var rateLimited map[string]time.Time + if r.backoff > 0 { + rateLimited, err = r.store.RateLimitedVideoIDs(ctx, r.userID) + if err != nil { + return stats, fmt.Errorf("runner: load rate-limited videos: %w", err) + } + } + subs, err := r.src.ListSubscriptions(ctx, r.userID) if err != nil { return stats, fmt.Errorf("runner: list subscriptions: %w", err) @@ -142,6 +182,15 @@ func (r *Runner) RunOnce(ctx context.Context) (Stats, error) { continue } + // Still inside the rate-limit backoff window: skip without fetching, so + // we don't re-hit a caption endpoint that just 429'd us. After the window + // expires the video falls through and is retried normally. + if at, ok := rateLimited[id]; ok && r.now().Sub(at) < r.backoff { + stats.SkippedRateLimited++ + r.log.Info("skipped video (rate-limited, backing off)", "video", v.ProviderVideoID, "title", v.Title) + continue + } + if fetchDelay > 0 { time.Sleep(fetchDelay) } @@ -152,11 +201,28 @@ func (r *Runner) RunOnce(ctx context.Context) (Stats, error) { continue } switch { + case res.Skipped && res.TranscriptSource == string(domain.SourceRateLimited): + // Fresh 429 this pass: persist rate_limited (stamps the backoff clock) + // so the next pass skips it until the window expires. + stats.SkippedRateLimited++ + if err := r.store.SetTranscriptStatus(ctx, r.userID, id, "rate_limited"); err != nil { + errs = append(errs, fmt.Errorf("set rate_limited status %q: %w", v.ProviderVideoID, err)) + stats.Errors++ + } + r.log.Info("skipped video (rate-limited)", "video", v.ProviderVideoID, "title", v.Title) case res.Skipped: stats.SkippedNoText++ + if err := r.store.SetTranscriptStatus(ctx, r.userID, id, "none"); err != nil { + errs = append(errs, fmt.Errorf("set none status %q: %w", v.ProviderVideoID, err)) + stats.Errors++ + } r.log.Info("skipped video (no transcript)", "video", v.ProviderVideoID, "title", v.Title) case res.Summary != nil: stats.Summarized++ + if err := r.store.SetTranscriptStatus(ctx, r.userID, id, "fetched"); err != nil { + errs = append(errs, fmt.Errorf("set fetched status %q: %w", v.ProviderVideoID, err)) + stats.Errors++ + } // In manual mode the video was processed because it was queued; // clear the flag so it is not re-summarized and the UI drops the // "Queued" chip. (Auto mode never sets the flag.) @@ -184,7 +250,8 @@ func (r *Runner) Loop(ctx context.Context, interval time.Duration) error { r.log.Info("run pass complete", "candidates", stats.Candidates, "summarized", stats.Summarized, "skipped_seen", stats.SkippedSeen, "skipped_no_text", stats.SkippedNoText, - "skipped_manual", stats.SkippedManual, "errors", stats.Errors) + "skipped_manual", stats.SkippedManual, "skipped_rate_limited", stats.SkippedRateLimited, + "errors", stats.Errors) if err != nil { r.log.Warn("run pass had errors", "err", err) } diff --git a/internal/runner/runner_test.go b/internal/runner/runner_test.go index eb08480..97527d2 100644 --- a/internal/runner/runner_test.go +++ b/internal/runner/runner_test.go @@ -5,6 +5,7 @@ import ( "io" "log/slog" "testing" + "time" "github.com/stretchr/testify/require" @@ -43,11 +44,13 @@ func (f *fakeSource) FetchTranscript(_ context.Context, v domain.Video) (domain. // auto controls the summarization mode; requested is the manual-mode queue keyed // by store id; cleared records the ids whose queue flag the runner reset. type fakeStore struct { - seen map[string]bool - upserted []domain.Video - auto bool - requested map[string]bool - cleared []string + seen map[string]bool + upserted []domain.Video + auto bool + requested map[string]bool + cleared []string + rateLimited map[string]time.Time // id -> when 429'd (seeds the backoff window) + statuses map[string]string // id -> last SetTranscriptStatus value } func (f *fakeStore) UpsertVideo(_ context.Context, v domain.Video) (string, error) { @@ -80,6 +83,22 @@ func (f *fakeStore) ClearSummarizeRequested(_ context.Context, _, videoID string return nil } +func (f *fakeStore) RateLimitedVideoIDs(_ context.Context, _ string) (map[string]time.Time, error) { + cp := make(map[string]time.Time, len(f.rateLimited)) + for k, v := range f.rateLimited { + cp[k] = v + } + return cp, nil +} + +func (f *fakeStore) SetTranscriptStatus(_ context.Context, _, videoID, status string) error { + if f.statuses == nil { + f.statuses = map[string]string{} + } + f.statuses[videoID] = status + return nil +} + type fakeSummarizer struct{} func (fakeSummarizer) Summarize(_ context.Context, v domain.Video, _ domain.Transcript) (domain.Summary, error) { @@ -207,6 +226,62 @@ func TestRunOnce_ManualMode_ProcessesRequested(t *testing.T) { require.Equal(t, []string{"id-v1"}, st.cleared, "the queue flag is cleared after summarizing") } +// noFetchSource fails the test if a transcript fetch happens — used to prove the +// runner skips a rate-limited video before touching the caption endpoint. +type noFetchSource struct{ *fakeSource } + +func (noFetchSource) FetchTranscript(context.Context, domain.Video) (domain.Transcript, error) { + panic("FetchTranscript must not be called for a rate-limited video within the backoff window") +} + +func TestRunOnce_SkipsRateLimitedWithinBackoff(t *testing.T) { + base := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC) + src := &fakeSource{ + subs: []domain.Subscription{sub("chan1", "Channel One")}, + videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1")}}, + } + // v1 was rate-limited 5m ago; backoff is 1h, so it is still inside the window. + st := &fakeStore{ + seen: map[string]bool{}, + auto: true, + rateLimited: map[string]time.Time{"id-v1": base.Add(-5 * time.Minute)}, + } + eng := usecase.NewEngine(noFetchSource{src}, fakeSummarizer{}, &recordingSink{}) + r := runner.New(noFetchSource{src}, st, eng, testUser, quietLogger(), + runner.WithBackoff(time.Hour), runner.WithClock(func() time.Time { return base })) + + stats, err := r.RunOnce(context.Background()) + require.NoError(t, err) + require.Equal(t, 1, stats.SkippedRateLimited, "still throttled -> skipped") + require.Equal(t, 0, stats.Summarized) + require.Empty(t, st.statuses, "no status write: the engine was never invoked") +} + +func TestRunOnce_RetriesRateLimitedAfterBackoff(t *testing.T) { + base := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC) + src := &fakeSource{ + subs: []domain.Subscription{sub("chan1", "Channel One")}, + videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1")}}, + } + // v1 was rate-limited 2h ago; backoff is 1h, so the window has expired. + st := &fakeStore{ + seen: map[string]bool{}, + auto: true, + rateLimited: map[string]time.Time{"id-v1": base.Add(-2 * time.Hour)}, + } + sink := &recordingSink{} + eng := usecase.NewEngine(src, fakeSummarizer{}, sink) + r := runner.New(src, st, eng, testUser, quietLogger(), + runner.WithBackoff(time.Hour), runner.WithClock(func() time.Time { return base })) + + stats, err := r.RunOnce(context.Background()) + require.NoError(t, err) + require.Equal(t, 0, stats.SkippedRateLimited, "window expired -> not skipped") + require.Equal(t, 1, stats.Summarized, "the video is retried and summarized") + require.Len(t, sink.delivered, 1) + require.Equal(t, "fetched", st.statuses["id-v1"], "status advances to fetched on success") +} + func TestRunOnce_UpsertsEveryCandidate(t *testing.T) { src := &fakeSource{ subs: []domain.Subscription{sub("chan1", "Channel One")},