mathiasandClaude Opus 4.8
beeb5bc31b
A user waiting on a Summarize click shared the per-IP caption gate equally with the background firehose, so on a busy IP the click was slow or 429'd. Add a context-marked priority lane: the web path (engineProcessor.ProcessVideo) marks its context foreground; the gate serves foreground immediately while background fetches yield until no foreground is pending. Threaded via a context value (no new signatures) + a process-wide foregroundPending counter. Clicks are rare, so the background barely loses throughput; the waiting human gets the cleaner slot. Drops the credentials probe: ADR-010 and captions.go already settle it — the timedtext/InnerTube path rejects authenticated requests and the OAuth token does not authenticate it anyway, so auth cannot help and can hurt. Documented in ADR-026 rather than built. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
85 lines
3.4 KiB
Go
85 lines
3.4 KiB
Go
package youtube
|
|
|
|
import (
|
|
"context"
|
|
"sync/atomic"
|
|
"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)
|
|
}
|
|
|
|
// foregroundPending counts in-flight foreground (user-initiated) caption fetches.
|
|
// The background sweep yields the gate while this is non-zero so a human waiting
|
|
// on a click gets the next slot — and, on a near-throttled IP, the pre-429 window
|
|
// — instead of competing equally with the firehose (ADR-026, Pillar A). Clicks are
|
|
// rare and bursty, so background barely notices; the win to the click is large.
|
|
var foregroundPending atomic.Int64
|
|
|
|
// fgCtxKey marks a context as foreground (user-initiated). Unexported; set via
|
|
// ForegroundContext and read via isForeground so only this package owns the key.
|
|
type fgCtxKey struct{}
|
|
|
|
// ForegroundContext marks ctx as a user-initiated (foreground) fetch so the gate
|
|
// gives it priority. The web "Summarize"/paste/retry path wraps its context with
|
|
// this; the background scheduler leaves it unset.
|
|
func ForegroundContext(ctx context.Context) context.Context {
|
|
return context.WithValue(ctx, fgCtxKey{}, true)
|
|
}
|
|
|
|
func isForeground(ctx context.Context) bool {
|
|
v, _ := ctx.Value(fgCtxKey{}).(bool)
|
|
return v
|
|
}
|
|
|
|
// fgYieldPoll is how often a background waiter re-checks whether a foreground
|
|
// fetch is still pending. Short enough to feel immediate, long enough not to spin.
|
|
const fgYieldPoll = 200 * time.Millisecond
|
|
|
|
// 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.
|
|
//
|
|
// Foreground (user-initiated) fetches take priority: they register as pending and
|
|
// acquire a token immediately. Background fetches first yield — they wait until no
|
|
// foreground fetch is pending — so a live click is never stuck behind the
|
|
// background sweep and gets the cleaner slot against the per-IP limit (ADR-026).
|
|
func WaitFetchGate(ctx context.Context) error {
|
|
if isForeground(ctx) {
|
|
foregroundPending.Add(1)
|
|
defer foregroundPending.Add(-1)
|
|
return globalFetchGate.Wait(ctx)
|
|
}
|
|
|
|
// Background: defer to any pending foreground fetch before taking a token.
|
|
for foregroundPending.Load() > 0 {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-time.After(fgYieldPoll):
|
|
}
|
|
}
|
|
return globalFetchGate.Wait(ctx)
|
|
}
|