Affected by 
Documentation
¶
Overview ¶
Package limiter provides a distributed, per-key concurrency governor for outbound model-provider calls. The shared finite resource is the model provider (its request/concurrency budget), so concurrency is capped at the model-client layer — keyed by model ID — rather than at the asynq queue layer (queue weights are scheduling priority, not throttling).
The Redis implementation is a self-healing distributed semaphore built on a sorted set: each held slot is a ZSET member (unique token) scored by its lease expiry. Acquire atomically prunes expired leases, counts live holders, and admits a new one only while under the limit. A background heartbeat refreshes the lease so long calls keep their slot; a crashed holder's lease simply expires and is reclaimed. Every backend error fails OPEN (the call is allowed) so a limiter/Redis outage can never halt model traffic.
Index ¶
- func Gate(ctx context.Context, modelID string) func()
- func GateN(ctx context.Context, modelID string, modelLimit int) func()
- func GateNamedN(ctx context.Context, modelID, modelName string, modelLimit int) func()
- func SetGlobalLimit(limit int)
- func SetGovernor(l ModelConcurrencyLimiter, limit int)
- type ModelConcurrencyLimiter
- type RuntimeStat
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Gate ¶
Gate acquires a per-model concurrency slot using the process-wide default limit. Equivalent to GateN(ctx, modelID, 0).
func GateN ¶
GateN acquires a per-model concurrency slot when the call is a background task (see types.IsBackgroundTask) and a governor is installed. modelLimit is the model's own configured cap; a value <= 0 means "fall back to the process-wide default" (governorN). It returns a release func that is ALWAYS safe to call: on the passthrough / fail-open paths it is a cheap no-op. The gate never blocks a call permanently — a limiter/Redis outage or a cancelled context fails open.
func GateNamedN ¶
func SetGlobalLimit ¶
func SetGlobalLimit(limit int)
SetGlobalLimit updates ONLY the process-wide default per-model limit, leaving the installed limiter backend intact. Used by the system-settings runtime bridge so an operator can retune model.max_concurrency without a restart. A non-positive value disables the default (models that carry their own MaxConcurrency still honour it).
func SetGovernor ¶
func SetGovernor(l ModelConcurrencyLimiter, limit int)
SetGovernor installs the process-wide background concurrency governor and the default per-model limit. Passing a nil limiter or a non-positive limit disables governance (all calls pass through). Safe to call at startup.
Types ¶
type ModelConcurrencyLimiter ¶
type ModelConcurrencyLimiter interface {
// Acquire blocks until a slot for key is available or ctx is done. It
// returns a release func that MUST be invoked to free the slot. On any
// backend error (or ctx cancellation) it fails open: release is a no-op and
// err is nil, so callers proceed without a slot rather than dropping the
// call.
Acquire(ctx context.Context, key string, limit int) (release func(), err error)
}
ModelConcurrencyLimiter caps the number of concurrent in-flight calls per key (typically a model ID) across all processes sharing the same backend.
func NewLocalLimiter ¶
func NewLocalLimiter() ModelConcurrencyLimiter
NewLocalLimiter builds an in-process per-key concurrency limiter.
func NewRedisLimiter ¶
func NewRedisLimiter(rdb *redis.Client) ModelConcurrencyLimiter
NewRedisLimiter builds a distributed limiter backed by rdb. A nil client yields a limiter that always fails open.
type RuntimeStat ¶
type RuntimeStat struct {
ModelID string `json:"model_id" extensions:"x-agent"`
Name string `json:"name" extensions:"x-agent"`
Active int64 `json:"active" extensions:"x-agent"`
Waiting int64 `json:"waiting" extensions:"x-agent"`
Limit int `json:"limit" extensions:"x-agent"`
}
RuntimeStat is a point-in-time view of a model semaphore. Active is cluster-wide for the Redis backend and process-local in Lite mode. Waiting is deliberately process-local: waiters block in application processes and are not represented in Redis.
func RuntimeStats ¶
func RuntimeStats(ctx context.Context) ([]RuntimeStat, bool, error)
RuntimeStats returns the semaphores observed by this process. Redis-backed limiters report cluster-wide active holders; waiting callers remain local to this instance by design.
Source Files