From f6623afd412ff1da1de251aebf4f6a848bcdee21 Mon Sep 17 00:00:00 2001 From: Mathias Date: Fri, 5 Jun 2026 23:39:39 +0200 Subject: [PATCH] feat(serve): in-process scheduled discovery for all users (ADR-018) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 0's "returns and reads in >=2 weeks" gate can't be met while discovery is host-side manual (`tapir run`): a newly onboarded user sees an empty list and never comes back. Make Tapir watch on its own. cmdServe launches a background goroutine (when TAPIR_DISCOVERY_INTERVAL > 0) that runs a discovery pass for ALL users on that cadence: enumerate via the un-RLS'd ListAllUsers, then run each user's pass through the EXISTING runner.Runner — the only new code is the per-user loop, not a new scheduler. Run-once-on-startup then ticked; ctx-cancelled on SIGTERM; per-user failures (including buildUserRunner errors) are logged and skipped so one bad user never aborts the rest. interval <= 0 disables it entirely (dev/tests). buildUserRunner binds each runner to that user's own YouTube refresh token (web.YouTubeTokenRef) — the Stage-1 per-tenant ref — reusing buildProcessor's engine wiring. SetFetchRate is also wired in cmdServe so the click-path shares the gate. SINGLE-REPLICA is now load-bearing: the loop lives in the web process, so >1 replica double-runs discovery (429s + duplicate work). Documented in cmdServe and warned at startup. Co-Authored-By: Claude Opus 4.8 (1M context) --- cmd/tapir/main.go | 28 +++++++ cmd/tapir/scheduler.go | 135 ++++++++++++++++++++++++++++++++ cmd/tapir/scheduler_test.go | 149 ++++++++++++++++++++++++++++++++++++ 3 files changed, 312 insertions(+) create mode 100644 cmd/tapir/scheduler.go create mode 100644 cmd/tapir/scheduler_test.go diff --git a/cmd/tapir/main.go b/cmd/tapir/main.go index 5186f1e..fc95766 100644 --- a/cmd/tapir/main.go +++ b/cmd/tapir/main.go @@ -165,6 +165,11 @@ func cmdServe(ctx context.Context, log *slog.Logger) error { } defer st.Close() + // Process-wide caption-fetch rate gate (ADR-014 item 2): the web click-path + // and the scheduled-discovery runners share one per-egress-IP limiter so they + // cannot collectively trip 429s. Must be set before either path fetches. + youtube.SetFetchRate(cfg.FetchRate) + // Auth seam (handlers depend on web.Auth only). With Dex configured // (TAPIR_OIDC_ISSUER set) serve uses real OIDC login — any Dex subject may // authenticate, then registers a tapir user (ADR-012); otherwise it falls @@ -234,6 +239,29 @@ func cmdServe(ctx context.Context, log *slog.Logger) error { log.Info("web summarization is queue-only (incomplete engine config)") } + // In-process scheduled discovery (ADR-018): when enabled, a background + // goroutine runs a discovery pass for ALL users on TAPIR_DISCOVERY_INTERVAL, + // reusing the per-user runner.Runner. Cancelled by the same ctx as the server. + // + // SINGLE-REPLICA ASSUMPTION (load-bearing): this loop lives in the web process. + // Running serve at >1 replica would make every replica fetch every user in + // parallel — duplicate work and self-inflicted 429s. replicas: 1 is required in + // the deployment manifest; scaling up needs a CronJob or leader election first. + if cfg.DiscoveryInterval > 0 { + log.Info("scheduled discovery enabled", "interval", cfg.DiscoveryInterval, "fetch_rate", cfg.FetchRate) + log.Warn("scheduled discovery assumes a SINGLE replica — running serve at >1 replica double-runs discovery (ADR-018)") + runUser := func(ctx context.Context, userID string) (runner.Stats, error) { + r, err := buildUserRunner(cfg, st, secretStore, userID, log) + if err != nil { + return runner.Stats{}, err + } + return r.RunOnce(ctx) + } + go runScheduler(ctx, cfg.DiscoveryInterval, st, runUser, log) + } else { + log.Info("scheduled discovery disabled (TAPIR_DISCOVERY_INTERVAL unset or 0)") + } + srv := &http.Server{ Addr: cfg.HTTPAddr, Handler: app.Router(), diff --git a/cmd/tapir/scheduler.go b/cmd/tapir/scheduler.go new file mode 100644 index 0000000..2525166 --- /dev/null +++ b/cmd/tapir/scheduler.go @@ -0,0 +1,135 @@ +package main + +import ( + "context" + "fmt" + "log/slog" + "time" + + "gitea.d-ma.be/mathias/tapir/internal/adapters/llm" + "gitea.d-ma.be/mathias/tapir/internal/adapters/store" + "gitea.d-ma.be/mathias/tapir/internal/adapters/summarizer" + "gitea.d-ma.be/mathias/tapir/internal/adapters/youtube" + "gitea.d-ma.be/mathias/tapir/internal/config" + "gitea.d-ma.be/mathias/tapir/internal/ports" + "gitea.d-ma.be/mathias/tapir/internal/runner" + "gitea.d-ma.be/mathias/tapir/internal/usecase" + "gitea.d-ma.be/mathias/tapir/internal/web" +) + +// buildUserRunner constructs a runner.Runner for one user, reusing the same +// engine wiring as buildProcessor but bound to that user's own YouTube refresh +// token (web.YouTubeTokenRef(userID)) — the Stage-1 per-tenant ref, not the +// Stage-0 single ref. It returns an error (not nil) when the global config can't +// support live summarization (gateway, YouTube client creds, secrets file), so +// the scheduler can skip that user gracefully. A user who simply hasn't connected +// YouTube yet builds fine here; their token ref fails to resolve at RunOnce time, +// surfacing as a per-user error the scheduler logs and skips. +func buildUserRunner(cfg config.Config, st *store.Store, secretStore ports.SecretStore, userID string, log *slog.Logger) (*runner.Runner, error) { + if cfg.GatewayURL == "" || cfg.YTClientID == "" || cfg.YTClientSecret == "" || cfg.SecretsFile == "" { + return nil, fmt.Errorf("buildUserRunner: incomplete summarization config (gateway, youtube credentials, secrets file)") + } + + src := youtube.New(youtube.Config{ + ClientID: cfg.YTClientID, + ClientSecret: cfg.YTClientSecret, + TokenSecretRef: web.YouTubeTokenRef(userID), + PreferredLanguages: []string{"en"}, + }, secretStore) + + primary := summarizer.Endpoint{ + Client: llm.New(cfg.GatewayURL, cfg.GatewayKey, cfg.SummarizerModel, cfg.SummarizerTimeout), + Provider: "local", + Model: cfg.SummarizerModel, + } + engine := usecase.NewEngine(src, summarizer.New(primary, nil), st) + + return runner.New(src, st, engine, userID, log, runner.WithBackoff(cfg.FetchBackoff)), nil +} + +// userLister enumerates every registered user. *store.Store satisfies it via +// ListAllUsers. A small local interface keeps the scheduler testable with a fake. +type userLister interface { + ListAllUsers(ctx context.Context) ([]store.UserIdentity, error) +} + +// runDiscoveryPass runs one discovery pass for every user. runUser performs a +// single user's pass (production: build a runner and RunOnce). Per-user failures +// — including a buildUserRunner error or a RunOnce error — are logged and skipped +// so one bad user, channel, or video never aborts the others (ADR-018 failure +// isolation). Returns the stats summed across users. +func runDiscoveryPass( + ctx context.Context, + lister userLister, + runUser func(context.Context, string) (runner.Stats, error), + log *slog.Logger, +) runner.Stats { + users, err := lister.ListAllUsers(ctx) + if err != nil { + log.Error("scheduler: list users failed", "err", err) + return runner.Stats{} + } + + log.Info("scheduler: starting discovery pass", "users", len(users)) + var total runner.Stats + for _, u := range users { + if ctx.Err() != nil { + break // shutting down: stop enumerating + } + stats, err := runUser(ctx, u.UserID) + total = sumStats(total, stats) + if err != nil { + log.Warn("scheduler: user discovery pass had errors", "user", u.UserID, "err", err) + } + } + log.Info("scheduler: pass complete", + "candidates", total.Candidates, "summarized", total.Summarized, + "skipped_seen", total.SkippedSeen, "skipped_no_text", total.SkippedNoText, + "skipped_manual", total.SkippedManual, "skipped_rate_limited", total.SkippedRateLimited, + "errors", total.Errors) + return total +} + +// runScheduler runs a discovery pass on startup, then once every interval until +// ctx is cancelled (pod SIGTERM exits the loop cleanly). A non-positive interval +// disables scheduling entirely (no startup pass) so dev/tests never auto-fetch. +// It reuses the existing runner.Runner via runUser — the only new behaviour over +// runner.Loop is iterating all users per tick (ADR-018). +func runScheduler( + ctx context.Context, + interval time.Duration, + lister userLister, + runUser func(context.Context, string) (runner.Stats, error), + log *slog.Logger, +) { + if interval <= 0 { + return // disabled + } + + runDiscoveryPass(ctx, lister, runUser, log) + + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + runDiscoveryPass(ctx, lister, runUser, log) + } + } +} + +// sumStats adds two passes' stats field-wise, so runDiscoveryPass can report a +// per-tick aggregate across all users. +func sumStats(a, b runner.Stats) runner.Stats { + return runner.Stats{ + Candidates: a.Candidates + b.Candidates, + Summarized: a.Summarized + b.Summarized, + SkippedSeen: a.SkippedSeen + b.SkippedSeen, + SkippedNoText: a.SkippedNoText + b.SkippedNoText, + SkippedManual: a.SkippedManual + b.SkippedManual, + SkippedRateLimited: a.SkippedRateLimited + b.SkippedRateLimited, + Errors: a.Errors + b.Errors, + } +} diff --git a/cmd/tapir/scheduler_test.go b/cmd/tapir/scheduler_test.go new file mode 100644 index 0000000..97fd536 --- /dev/null +++ b/cmd/tapir/scheduler_test.go @@ -0,0 +1,149 @@ +package main + +import ( + "context" + "errors" + "io" + "log/slog" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "gitea.d-ma.be/mathias/tapir/internal/adapters/store" + "gitea.d-ma.be/mathias/tapir/internal/runner" +) + +func quietLog() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// fakeLister returns a fixed user set (or an error) for the scheduler under test. +type fakeLister struct { + users []store.UserIdentity + err error +} + +func (f fakeLister) ListAllUsers(context.Context) ([]store.UserIdentity, error) { + return f.users, f.err +} + +// countingRunUser records how many passes each user got, optionally failing for +// specific users, under a mutex so it is safe across the scheduler goroutine. +type countingRunUser struct { + mu sync.Mutex + calls map[string]int + failFor map[string]bool +} + +func newCountingRunUser(failFor ...string) *countingRunUser { + c := &countingRunUser{calls: map[string]int{}, failFor: map[string]bool{}} + for _, u := range failFor { + c.failFor[u] = true + } + return c +} + +func (c *countingRunUser) run(_ context.Context, userID string) (runner.Stats, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls[userID]++ + if c.failFor[userID] { + return runner.Stats{Errors: 1}, errors.New("boom") + } + return runner.Stats{Summarized: 1}, nil +} + +func (c *countingRunUser) count(userID string) int { + c.mu.Lock() + defer c.mu.Unlock() + return c.calls[userID] +} + +func (c *countingRunUser) total() int { + c.mu.Lock() + defer c.mu.Unlock() + n := 0 + for _, v := range c.calls { + n += v + } + return n +} + +func usersN(ids ...string) []store.UserIdentity { + out := make([]store.UserIdentity, len(ids)) + for i, id := range ids { + out[i] = store.UserIdentity{UserID: id, DexSubject: "dex|" + id} + } + return out +} + +func TestDiscoveryPassRunsEveryUserOnce(t *testing.T) { + lister := fakeLister{users: usersN("a", "b", "c")} + rc := newCountingRunUser() + + stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog()) + + require.Equal(t, 1, rc.count("a")) + require.Equal(t, 1, rc.count("b")) + require.Equal(t, 1, rc.count("c")) + require.Equal(t, 3, stats.Summarized, "stats are summed across users") +} + +func TestDiscoveryPassOneUserFailureDoesNotStopOthers(t *testing.T) { + lister := fakeLister{users: usersN("a", "b", "c")} + rc := newCountingRunUser("b") // user b's pass errors + + stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog()) + + require.Equal(t, 1, rc.count("a")) + require.Equal(t, 1, rc.count("b")) + require.Equal(t, 1, rc.count("c"), "a failing user must not abort the rest") + require.Equal(t, 2, stats.Summarized) // a + c + require.Equal(t, 1, stats.Errors) // b +} + +func TestDiscoveryPassListerErrorIsContained(t *testing.T) { + lister := fakeLister{err: errors.New("db down")} + rc := newCountingRunUser() + + stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog()) + + require.Equal(t, 0, rc.total(), "no users enumerated → no passes") + require.Equal(t, runner.Stats{}, stats) +} + +func TestSchedulerIntervalZeroDisablesEntirely(t *testing.T) { + lister := fakeLister{users: usersN("a", "b")} + rc := newCountingRunUser() + + runScheduler(context.Background(), 0, lister, rc.run, quietLog()) + + require.Equal(t, 0, rc.total(), "interval 0 must not run even a startup pass") +} + +func TestSchedulerRunsStartupPassThenStopsOnCancel(t *testing.T) { + lister := fakeLister{users: usersN("a", "b", "c")} + rc := newCountingRunUser() + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + // A long interval so only the startup pass runs before we cancel. + runScheduler(ctx, time.Hour, lister, rc.run, quietLog()) + close(done) + }() + + // The startup pass is synchronous at the top of runScheduler; once all three + // users have a pass it has completed and the loop is parked on the ticker. + require.Eventually(t, func() bool { return rc.total() == 3 }, time.Second, 5*time.Millisecond) + + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("scheduler did not exit after ctx cancel") + } + require.Equal(t, 3, rc.total(), "no extra passes fired between startup and cancel") +}