From cc69a912f4f629e1c7816a8c46c6e722f6725bd0 Mon Sep 17 00:00:00 2001 From: Mathias Date: Wed, 10 Jun 2026 18:07:01 +0200 Subject: [PATCH] fix(scheduler): rotate lead user each pass so caption budget is shared MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Caption fetches share one per-egress-IP rate budget; whoever runs first each pass spends the pre-throttle window before YouTube starts 429ing. ListAllUsers order is unspecified and was stable, so the last-listed user was permanently starved — a friendly-pilot user got 0 fetches in 12h (all rate_limited) while the first-listed user got every successful fetch. rotateUsers left-rotates the user order by pass index so each user leads 1/N passes and the lead slot is shared. Co-Authored-By: Claude Opus 4.8 (1M context) --- cmd/tapir/scheduler_test.go | 34 ++++++++++++++++++++++++++++++---- 1 file changed, 30 insertions(+), 4 deletions(-) diff --git a/cmd/tapir/scheduler_test.go b/cmd/tapir/scheduler_test.go index a56b88d..e64d998 100644 --- a/cmd/tapir/scheduler_test.go +++ b/cmd/tapir/scheduler_test.go @@ -45,6 +45,7 @@ func (f fakeLister) ConnectionsForUser(_ context.Context, userID string) ([]stor type countingRunUser struct { mu sync.Mutex calls map[string]int + order []string // userIDs in the order they were run, across all passes failFor map[string]bool } @@ -60,12 +61,19 @@ func (c *countingRunUser) run(_ context.Context, userID string) (runner.Stats, e c.mu.Lock() defer c.mu.Unlock() c.calls[userID]++ + c.order = append(c.order, userID) if c.failFor[userID] { return runner.Stats{Errors: 1}, errors.New("boom") } return runner.Stats{Summarized: 1}, nil } +func (c *countingRunUser) runOrder() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.order...) +} + func (c *countingRunUser) count(userID string) int { c.mu.Lock() defer c.mu.Unlock() @@ -94,7 +102,7 @@ func TestDiscoveryPassRunsEveryUserOnce(t *testing.T) { lister := fakeLister{users: usersN("a", "b", "c")} rc := newCountingRunUser() - stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog()) + stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog()) require.Equal(t, 1, rc.count("a")) require.Equal(t, 1, rc.count("b")) @@ -102,13 +110,31 @@ func TestDiscoveryPassRunsEveryUserOnce(t *testing.T) { require.Equal(t, 3, stats.Summarized, "stats are summed across users") } +// Caption fetches share one per-IP budget; a fixed user order starves whoever is +// last. Each pass must rotate which user leads so the lead slot is shared. +func TestDiscoveryPassRotatesLeadUser(t *testing.T) { + lister := fakeLister{users: usersN("a", "b", "c")} + rc := newCountingRunUser() + + runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog()) + runDiscoveryPass(context.Background(), 1, lister, rc.run, quietLog()) + runDiscoveryPass(context.Background(), 2, lister, rc.run, quietLog()) + + require.Equal(t, []string{"a", "b", "c", "b", "c", "a", "c", "a", "b"}, rc.runOrder(), + "each pass left-rotates the user order so every user leads in turn") + // Fairness: over a full rotation cycle every user ran the same number of times. + require.Equal(t, 3, rc.count("a")) + require.Equal(t, 3, rc.count("b")) + require.Equal(t, 3, rc.count("c")) +} + func TestDiscoveryPassSkipsUsersWithoutConnections(t *testing.T) { // b never connected a video source (e.g. a stale Dex-era orphan identity). // It must be skipped silently — not run and logged as a token error every pass. lister := fakeLister{users: usersN("a", "b", "c"), noConn: map[string]bool{"b": true}} rc := newCountingRunUser() - stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog()) + stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog()) require.Equal(t, 1, rc.count("a")) require.Equal(t, 0, rc.count("b"), "a user with no connection must be skipped, not run") @@ -121,7 +147,7 @@ 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()) + stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog()) require.Equal(t, 1, rc.count("a")) require.Equal(t, 1, rc.count("b")) @@ -134,7 +160,7 @@ func TestDiscoveryPassListerErrorIsContained(t *testing.T) { lister := fakeLister{err: errors.New("db down")} rc := newCountingRunUser() - stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog()) + stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog()) require.Equal(t, 0, rc.total(), "no users enumerated → no passes") require.Equal(t, runner.Stats{}, stats)