fix(playback): resolve group-merged cap in async stream enforcer

The enforcer's LimitFunc read the raw users.max_streams column, which is 0
("inherit from group") for every standard account since migrations
20260702180000/20260702190000 moved the real cap into the Default Group.
The enforcer treats limit <= 0 as unlimited, so the async over-cap brain
never trimmed anyone on a default install — only per-process synchronous
admission held, leaving the cross-node backstop it was built for a no-op.

Extract admission's limit lookup into a shared SessionLimitProvider
(GetByID + access.EffectivePolicyForUser) and feed the enforcer through
it, so admission and the enforcer can never disagree about a user's
effective cap again.

Part of #306.
This commit is contained in:
CoffeeKnyte
2026-07-29 12:28:06 +00:00
parent 53c9313b57
commit 0cc706e066
2 changed files with 18 additions and 3 deletions
+5 -3
View File
@@ -1732,7 +1732,6 @@ func main() {
// revocations the edges enforce. Source is the authoritative monitoring
// picture — Redis for multi-node, the in-process session manager for
// integrated. Fails open on limit-lookup errors.
enforcerUserRepo := auth.NewUserRepository(deps.DB)
// Union of (a) the in-process session manager — authoritative for streams
// this node serves directly, integrated or not — and (b) the edge Redis
// records — authoritative for offloaded/edge-served streams. Deduped by
@@ -1750,12 +1749,15 @@ func main() {
if apiRedisClient != nil {
monitorSource = streammonitor.NewMultiSource(localSource, streammonitor.NewRedisSource(apiRedisClient))
}
// Resolve caps through the session manager so the enforcer and synchronous
// admission share one provider (installed by the API wiring) and can never
// disagree about a user's effective, group-merged cap.
streamenforcer.New(monitorSource, func(ctx context.Context, userID int) (int, error) {
u, err := enforcerUserRepo.GetByID(ctx, userID)
limits, err := sessionMgr.EffectiveLimits(ctx, userID)
if err != nil {
return 0, err
}
return u.MaxStreams, nil
return limits.MaxStreams, nil
}, streamRevocation, 0).Start(appCtx)
nodeURL := fmt.Sprintf("http://%s%s", nodeIdentity, cfg.Server.Listen)
+13
View File
@@ -569,6 +569,19 @@ func (m *SessionManager) RegisterReconstructedWithLimits(ctx context.Context, s
return s, nil
}
// EffectiveLimits resolves a user's admission limits through the installed
// limit provider, falling back to the manager defaults when none is set.
//
// Out-of-band consumers — notably the async stream enforcer — must resolve caps
// through here rather than reading users.max_streams directly: that column is 0
// ("inherit from group") for standard accounts, which an enforcer treating
// limit <= 0 as unlimited would read as "no cap". Going through the same
// provider synchronous admission uses means the two can never disagree, and the
// lookup is late-bound so it sees whatever provider the API wiring installed.
func (m *SessionManager) EffectiveLimits(ctx context.Context, userID int) (SessionLimits, error) {
return m.limitsForUser(ctx, userID)
}
func (m *SessionManager) limitsForUser(ctx context.Context, userID int) (SessionLimits, error) {
m.mu.RLock()
provider := m.limitProvider