fix(scheduler): rotate over connected users only for true fair share
The lead-user rotation rotated the full ListAllUsers set, so a connectionless orphan identity ate a rotation slot — collapsing onto the next real user and skewing the lead share (two real users got 2/3 vs 1/3 instead of 50/50). Filter to connected users BEFORE rotating so the rotation is over exactly the users that consume the caption budget. A dead identity can no longer skew fairness. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
+24
-13
@@ -73,22 +73,17 @@ func runDiscoveryPass(
|
|||||||
return runner.Stats{}
|
return runner.Stats{}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Rotate who goes first each pass. Caption fetches share one per-egress-IP
|
// Keep only users with a video connection. A pass for a connectionless user
|
||||||
// rate budget (ADR-014); whoever runs first each pass spends the pre-throttle
|
// (e.g. a stale Dex-era orphan identity) only tries to resolve a token that
|
||||||
// window, so a FIXED user order permanently starves whoever is last (a new
|
// was never minted, logging a spurious "ref not found" every tick. Filtering
|
||||||
// pilot user got 0 fetches for 12h while the first-listed user got all of
|
// here — BEFORE rotation — also keeps fairness honest: rotation is over the
|
||||||
// them). Rotation gives every user the lead slot in turn.
|
// users that actually consume the caption budget, so a dead identity can't eat
|
||||||
users = rotateUsers(users, pass)
|
// a rotation slot and skew the lead share.
|
||||||
|
var connected []store.UserIdentity
|
||||||
log.Info("scheduler: starting discovery pass", "users", len(users))
|
|
||||||
var total runner.Stats
|
|
||||||
for _, u := range users {
|
for _, u := range users {
|
||||||
if ctx.Err() != nil {
|
if ctx.Err() != nil {
|
||||||
break // shutting down: stop enumerating
|
return runner.Stats{} // shutting down
|
||||||
}
|
}
|
||||||
// Skip users with no video connection. A discovery pass for them only
|
|
||||||
// attempts to resolve a token that was never minted, logging a spurious
|
|
||||||
// "ref not found" every tick (e.g. stale Dex-era orphan identities).
|
|
||||||
conns, err := lister.ConnectionsForUser(ctx, u.UserID)
|
conns, err := lister.ConnectionsForUser(ctx, u.UserID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Warn("scheduler: list connections failed", "user", u.UserID, "err", err)
|
log.Warn("scheduler: list connections failed", "user", u.UserID, "err", err)
|
||||||
@@ -98,6 +93,22 @@ func runDiscoveryPass(
|
|||||||
log.Debug("scheduler: skipping user with no video connections", "user", u.UserID)
|
log.Debug("scheduler: skipping user with no video connections", "user", u.UserID)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
connected = append(connected, u)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Rotate who goes first each pass. Caption fetches share one per-egress-IP
|
||||||
|
// rate budget (ADR-014); whoever runs first each pass spends the pre-throttle
|
||||||
|
// window, so a FIXED order permanently starves whoever is last (a new pilot
|
||||||
|
// user got 0 fetches for 12h while the first-listed user got all of them).
|
||||||
|
// Rotation over the connected set gives each real user the lead in turn.
|
||||||
|
connected = rotateUsers(connected, pass)
|
||||||
|
|
||||||
|
log.Info("scheduler: starting discovery pass", "users", len(connected))
|
||||||
|
var total runner.Stats
|
||||||
|
for _, u := range connected {
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
break // shutting down: stop enumerating
|
||||||
|
}
|
||||||
stats, err := runUser(ctx, u.UserID)
|
stats, err := runUser(ctx, u.UserID)
|
||||||
total = sumStats(total, stats)
|
total = sumStats(total, stats)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -128,6 +128,21 @@ func TestDiscoveryPassRotatesLeadUser(t *testing.T) {
|
|||||||
require.Equal(t, 3, rc.count("c"))
|
require.Equal(t, 3, rc.count("c"))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A connectionless orphan must not consume a rotation slot: rotation is over the
|
||||||
|
// connected users only, so two real users alternate the lead 50/50 even with a
|
||||||
|
// dead identity listed between them.
|
||||||
|
func TestDiscoveryPassRotationIgnoresConnectionlessUsers(t *testing.T) {
|
||||||
|
lister := fakeLister{users: usersN("a", "orphan", "c"), noConn: map[string]bool{"orphan": true}}
|
||||||
|
rc := newCountingRunUser()
|
||||||
|
|
||||||
|
runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
|
||||||
|
runDiscoveryPass(context.Background(), 1, lister, rc.run, quietLog())
|
||||||
|
|
||||||
|
require.Equal(t, []string{"a", "c", "c", "a"}, rc.runOrder(),
|
||||||
|
"only connected users rotate; the orphan never runs and never holds a slot")
|
||||||
|
require.Equal(t, 0, rc.count("orphan"))
|
||||||
|
}
|
||||||
|
|
||||||
func TestDiscoveryPassSkipsUsersWithoutConnections(t *testing.T) {
|
func TestDiscoveryPassSkipsUsersWithoutConnections(t *testing.T) {
|
||||||
// b never connected a video source (e.g. a stale Dex-era orphan identity).
|
// 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.
|
// It must be skipped silently — not run and logged as a token error every pass.
|
||||||
|
|||||||
Reference in New Issue
Block a user