diff --git a/cmd/silo/main.go b/cmd/silo/main.go index 5acca885..8784a9aa 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -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) diff --git a/internal/playback/session.go b/internal/playback/session.go index f1c40397..d52a1b1f 100644 --- a/internal/playback/session.go +++ b/internal/playback/session.go @@ -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