feat(youtube): process-wide caption-fetch rate gate (ADR-014 item 2)

ADR-014 item 2 — a single per-egress-IP rate gate shared by every caption
fetch — was specced but only per-video backoff (rate_limited_at) shipped.
Build the real gate now: it is load-bearing once ADR-018 puts auto-summarize
on an in-process schedule across multiple users (all fetches leave one pod's
egress IP, concurrently with live "Summarize" clicks — without a shared gate
that self-inflicts 429s every cycle).

globalFetchGate (golang.org/x/time/rate, default 2s/req burst 1) is consulted
in httpDo before every live outbound fetch — player, watch-page, timedtext —
so the scheduler runners and the web click-path serialise through one limiter
regardless of how many users/goroutines are upstream. The test seam
(a.transport != nil) skips the gate so fakes are not throttled.

TAPIR_FETCH_RATE (Go duration, default 2s, 0 = unlimited) wires SetFetchRate in
cmdRun; the existing per-video backoff stays as the complementary 429 handler.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-05 23:36:17 +02:00
co-authored by Claude Opus 4.8
parent 88294d38bc
commit f5021a8436
8 changed files with 176 additions and 1 deletions
+9
View File
@@ -235,6 +235,15 @@ func (a *Adapter) httpDo(ctx context.Context, client *http.Client, method, url s
for k, v := range headers {
req.Header.Set(k, v)
}
// Process-wide rate gate (ADR-014 item 2): every live caption fetch — player,
// watch-page, and timedtext baseUrl — passes the shared per-egress-IP limiter
// so the scheduler and the click-path cannot collectively trip 429s. Skipped
// when a.transport is set (the test seam) so fakes are not throttled.
if a.transport == nil {
if err := WaitFetchGate(ctx); err != nil {
return nil, 0, fmt.Errorf("fetch gate %s %s: %w", method, url, err)
}
}
resp, err := client.Do(req)
if err != nil {
return nil, 0, fmt.Errorf("%s %s: %w", method, url, err)
+37
View File
@@ -0,0 +1,37 @@
package youtube
import (
"context"
"time"
"golang.org/x/time/rate"
)
// globalFetchGate is the process-wide rate limiter for outbound timedtext/caption
// fetches. A single instance is shared by ALL Adapter instances (the scheduler's
// per-user runners + the web click-path) so they cannot collectively exceed the
// per-egress-IP cap. ADR-014 item 2: the gate serialises/limits concurrent caption
// fetches regardless of how many users or goroutines are upstream. The 429 is per
// IP, not per user — so the gate is process-wide, not per-withUser, not per-video.
//
// Default 2s/req (burst 1): the first fetch passes immediately, subsequent fetches
// are spaced at least 2s apart. Production overrides via SetFetchRate from config.
var globalFetchGate = rate.NewLimiter(rate.Every(2*time.Second), 1)
// SetFetchRate replaces the process-wide gate's rate with one token per interval.
// Call once at startup from config (TAPIR_FETCH_RATE). A non-positive interval
// installs an unlimited gate (rate.Inf) — used in dev/tests so nothing throttles.
func SetFetchRate(interval time.Duration) {
if interval <= 0 {
globalFetchGate = rate.NewLimiter(rate.Inf, 1)
return
}
globalFetchGate = rate.NewLimiter(rate.Every(interval), 1)
}
// WaitFetchGate blocks until the process-wide gate allows one timedtext fetch,
// respecting ctx cancellation. Called from httpDo before every live outbound
// caption fetch so the scheduler and the click-path share the same egress budget.
func WaitFetchGate(ctx context.Context) error {
return globalFetchGate.Wait(ctx)
}
+84
View File
@@ -0,0 +1,84 @@
package youtube
import (
"context"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/require"
"golang.org/x/time/rate"
)
// waitOn drives a test-scoped limiter the same way WaitFetchGate drives the
// global one, so these tests exercise the gate's behaviour without mutating the
// process-wide gate (which would pollute sibling tests / the click-path).
func waitOn(t *testing.T, lim *rate.Limiter) func(context.Context) error {
t.Helper()
return func(ctx context.Context) error { return lim.Wait(ctx) }
}
func TestFetchGateSerialisesConcurrentCallers(t *testing.T) {
const (
n = 5
interval = 10 * time.Millisecond
)
lim := rate.NewLimiter(rate.Every(interval), 1)
wait := waitOn(t, lim)
var (
inFlight, maxInFlight atomic.Int32
wg sync.WaitGroup
)
start := time.Now()
for i := 0; i < n; i++ {
wg.Add(1)
go func() {
defer wg.Done()
require.NoError(t, wait(context.Background()))
cur := inFlight.Add(1)
for {
old := maxInFlight.Load()
if cur <= old || maxInFlight.CompareAndSwap(old, cur) {
break
}
}
// Hold the "critical section" briefly so overlap would be observable.
time.Sleep(interval / 4)
inFlight.Add(-1)
}()
}
wg.Wait()
elapsed := time.Since(start)
require.Equal(t, int32(1), maxInFlight.Load(),
"the gate must admit at most one caller per interval — no overlap")
require.GreaterOrEqual(t, elapsed, time.Duration(n-1)*interval,
"N gated callers take at least (N-1)*interval wall time")
}
func TestFetchGateRespectsContextCancellation(t *testing.T) {
// A slow gate (1 token/hour, burst already spent) blocks; a cancelled ctx must
// unblock Wait with an error rather than hang.
lim := rate.NewLimiter(rate.Every(time.Hour), 1)
require.True(t, lim.Allow(), "spend the single burst token")
ctx, cancel := context.WithCancel(context.Background())
cancel()
require.Error(t, waitOn(t, lim)(ctx), "cancelled ctx must fail Wait, not block")
}
func TestSetFetchRateZeroIsUnlimited(t *testing.T) {
// Snapshot and restore the global so this test does not pollute the process.
prev := globalFetchGate
t.Cleanup(func() { globalFetchGate = prev })
SetFetchRate(0)
require.Equal(t, rate.Inf, globalFetchGate.Limit(), "0 interval = unlimited gate")
// An unlimited gate never blocks, even back-to-back.
for i := 0; i < 100; i++ {
require.NoError(t, WaitFetchGate(context.Background()))
}
}
+24
View File
@@ -65,6 +65,17 @@ type Config struct {
// expires the video is retried. Zero means "always retry" (no backoff).
FetchBackoff time.Duration
// FetchRate is the minimum interval between outbound caption fetches across the
// whole process — the shared per-egress-IP rate gate (ADR-014 item 2). It is
// the gate that makes auto-summarize-on-a-schedule safe: scheduler runners and
// the web click-path serialise through it. Zero = unlimited (dev/tests).
FetchRate time.Duration
// DiscoveryInterval, when > 0, makes `serve` run in-process scheduled discovery
// for ALL users on that cadence (ADR-018). Zero/unset = disabled, so dev and
// tests never auto-fetch. Single-replica assumption — see cmdServe.
DiscoveryInterval time.Duration
// HTTPAddr is the listen address for `tapir serve` (the Stage-0 web UI).
HTTPAddr string
@@ -97,6 +108,7 @@ const (
defaultOAuthRedirectAddr = "localhost:8080"
defaultHTTPAddr = ":8080"
defaultFetchBackoff = time.Hour
defaultFetchRate = 2 * time.Second
defaultPublicURL = "https://tapir.d-ma.be"
)
@@ -144,6 +156,18 @@ func Load() (Config, error) {
}
c.FetchBackoff = backoff
fetchRate, err := durationOr("TAPIR_FETCH_RATE", defaultFetchRate)
if err != nil {
return Config{}, err
}
c.FetchRate = fetchRate
discovery, err := durationOr("TAPIR_DISCOVERY_INTERVAL", 0)
if err != nil {
return Config{}, err
}
c.DiscoveryInterval = discovery
return c, nil
}