feat(runner): 429 backoff — skip still-throttled videos, persist status

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) <noreply@anthropic.com>
2026-06-03 22:56:07 +02:00
parent ce2fc62ef8
commit 40a614c8d4
3 changed files with 163 additions and 21 deletions
@@ -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)
 }

@@ -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)
 		}
@@ -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")},