diff --git a/.env.example b/.env.example index 3294e69..608e080 100644 --- a/.env.example +++ b/.env.example @@ -51,6 +51,18 @@ TAPIR_POLL_INTERVAL= # caption endpoint; after it expires the video is retried. 0 = always retry. # Go duration; default 1h. TAPIR_FETCH_BACKOFF= +# Minimum interval between outbound caption fetches across the WHOLE process — +# the shared per-egress-IP rate gate (ADR-014). Scheduler runners and the web +# "Summarize" click-path serialise through it so they cannot collectively trip +# 429s. Go duration; default 2s. 0 = unlimited (dev/tests). +TAPIR_FETCH_RATE= + +# --- scheduled discovery (tapir serve, ADR-018) --------------------------- +# When > 0, `serve` runs in-process discovery for ALL users on this cadence +# (e.g. 2h): one runner pass per user per tick, run-once-on-startup then ticked. +# Empty/0 = disabled (dev/tests never auto-fetch). SINGLE-REPLICA assumption — +# >1 replica double-runs discovery. Go duration. +TAPIR_DISCOVERY_INTERVAL= # --- invitations (tapir invite) ------------------------------------------- # Public base URL used to build the invite link `tapir invite ` prints. diff --git a/cmd/tapir/main.go b/cmd/tapir/main.go index 2c1d533..5186f1e 100644 --- a/cmd/tapir/main.go +++ b/cmd/tapir/main.go @@ -25,6 +25,7 @@ import ( "gitea.d-ma.be/mathias/tapir/internal/adapters/dex" "gitea.d-ma.be/mathias/tapir/internal/adapters/secrets" "gitea.d-ma.be/mathias/tapir/internal/adapters/store" + "gitea.d-ma.be/mathias/tapir/internal/adapters/youtube" "gitea.d-ma.be/mathias/tapir/internal/auth" "gitea.d-ma.be/mathias/tapir/internal/config" "gitea.d-ma.be/mathias/tapir/internal/runner" @@ -132,10 +133,15 @@ func cmdRun(ctx context.Context, log *slog.Logger) error { if engine == nil { return fmt.Errorf("run: incomplete summarization config (gateway, youtube credentials, secrets file)") } + // Process-wide caption-fetch rate gate (ADR-014 item 2): the batch path shares + // the same per-egress-IP limiter as the web click-path. + youtube.SetFetchRate(cfg.FetchRate) + 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, "fetch_backoff", cfg.FetchBackoff) + "gateway", cfg.GatewayURL, "poll_interval", cfg.PollInterval, "fetch_backoff", cfg.FetchBackoff, + "fetch_rate", cfg.FetchRate) return r.Loop(ctx, cfg.PollInterval) } diff --git a/go.mod b/go.mod index e7065b5..a9fa266 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/stretchr/testify v1.11.1 golang.org/x/crypto v0.45.0 golang.org/x/oauth2 v0.36.0 + golang.org/x/time v0.15.0 ) require ( diff --git a/go.sum b/go.sum index 7c52e63..9993d10 100644 --- a/go.sum +++ b/go.sum @@ -101,6 +101,8 @@ golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= diff --git a/internal/adapters/youtube/captions.go b/internal/adapters/youtube/captions.go index 975a934..f0b0762 100644 --- a/internal/adapters/youtube/captions.go +++ b/internal/adapters/youtube/captions.go @@ -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) diff --git a/internal/adapters/youtube/gate.go b/internal/adapters/youtube/gate.go new file mode 100644 index 0000000..08f52c4 --- /dev/null +++ b/internal/adapters/youtube/gate.go @@ -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) +} diff --git a/internal/adapters/youtube/gate_test.go b/internal/adapters/youtube/gate_test.go new file mode 100644 index 0000000..eeaefe8 --- /dev/null +++ b/internal/adapters/youtube/gate_test.go @@ -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())) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 6b132d2..e81b3e8 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -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 }