Allow model rate limits to select request classes

README.md

@@ -557,6 +557,29 @@ cap concurrency, request rate, or both:
   arrival instead of failing (NGINX `limit_req` without `nodelay`). Waits share
   the 256-request queue bound and the five-minute maximum; a request whose wait
   would exceed it still fails fast with `429` and `retry-after`.
+- `requestClasses`: optional non-empty array of `standard`, `interactive`, or
+  `foreground`. Only those classes use this limiter's rate budget, concurrency
+  slots, and waiting room; omitted means all classes. Selected classes share
+  one bucket. Other applicable model/alias limits and normal runtime capacity
+  still apply.
+
+For example, bind a cloud model's cost throttle to background work while chat
+keeps its normal priority and capacity:
+
+```json
+"rateLimit": {
+  "rate": "4.5/m",
+  "burst": 0,
+  "queue": true,
+  "requestClasses": ["standard"]
+}
+```
+
+Set this on the concrete cloud model to cover every alias that resolves to it.
+The limit follows the resolved model when routes change. A caller marked
+`standard`, including asynchronous inbox work, remains paced. Foreground and
+interactive requests bypass this particular limit without spending its budget.
+The rate counts model requests, not tokens or dollars.
 
 The limiter is the standard token-bucket used by NGINX `limit_req` and Envoy's
 local rate limiter: tokens refill continuously at one per `rateMs`, requests
@@ -571,7 +594,7 @@ same limited model draw from the same budget, and a direct model ID request is
 limited identically. Rate-limit rejections never poison target failure backoff
 or trigger model failover; the 429 surfaces directly with `retry-after`.
 
-Live state (active, queued, next-allowed time per limiter) is exposed as
+Live state (selected request classes, active, queued, next-allowed time per limiter) is exposed as
 `rateLimits` in `GET /gateway/routing`. Config reloads keep existing buckets
 for unchanged ids and prune removed ones.
 

docs/request-scheduling.md

@@ -106,6 +106,16 @@ the next token. Completed requests still consume their budget. Set an individual
 is exhausted. Limits and scheduling settings can be hot-reloaded without
 canceling active requests.
 
+An optional `rateLimit.requestClasses` array scopes a model or alias limit to
+specific existing priorities. For example, `requestClasses: ["standard"]` with
+`rate: "4.5/m"`, `burst: 0`, and `queue: true` spaces background calls without
+making foreground or interactive callers wait for that budget. The selected
+classes share the limiter; omitted means all three classes. Exempt requests
+neither reserve its concurrency slots nor spend its rate tokens. Every other
+applicable limit and runtime-capacity rule still applies, and priority within
+each compute group is unchanged. Config reloads preserve the spent bucket while
+updating which classes use it, including already waiting requests.
+
 Adaptive requests also wait through a needed cold start instead of receiving
 the earlier startup retry response. Model-loading admission checks priority
 before changing residency and yields before its next mutation when higher

src/rate-limit.mjs

@@ -21,6 +21,8 @@
 // the next conforming arrival instead (nginx `limit_req` with a burst queue
 // and no `nodelay`), bounded by the same maximum wait.
 
+import { REQUEST_CLASSES, normalizeSchedulingClass } from './request-class.mjs';
+
 const MAX_WAIT_MS = 5 * 60 * 1000;
 
 export const RATE_LIMIT_QUEUE_CAPACITY = 256;
@@ -54,6 +56,17 @@ export function normalizeRateLimit(value) {
     throw new Error('rateLimit must be a string like "20/m" or an object');
   }
   const settings = value.rateLimit ?? value;
+  const selection = {};
+  if (Object.hasOwn(settings, 'requestClasses')) {
+    if (
+      !Array.isArray(settings.requestClasses) ||
+      !settings.requestClasses.length ||
+      settings.requestClasses.some((requestClass) => !REQUEST_CLASSES.includes(requestClass))
+    ) {
+      throw new Error('rateLimit.requestClasses must be a non-empty array of standard, interactive, or foreground');
+    }
+    selection.requestClasses = REQUEST_CLASSES.filter((requestClass) => settings.requestClasses.includes(requestClass));
+  }
   if (settings.queue != null && typeof settings.queue !== 'boolean') {
     throw new Error(`rateLimit.queue must be a boolean: ${JSON.stringify(settings.queue)}`);
   }
@@ -77,7 +90,7 @@ export function normalizeRateLimit(value) {
       throw new Error('rateLimit.burst must be a non-negative integer');
     }
     burst = settings.burst ?? 0;
-    return { maxConcurrent, burst, rateMs, ...queue };
+    return { maxConcurrent, burst, rateMs, ...queue, ...selection };
   }
   const rate = settings.rate ?? settings.requestsPerMinute ?? settings.requests;
   let rateCount = null;
@@ -132,7 +145,8 @@ export function normalizeRateLimit(value) {
     // An explicit burst overrides this.
     burst: rateMs == null ? 0 : (burst ?? Math.max(0, Math.floor(rateCount) - 1)),
     rateMs,
-    ...queue
+    ...queue,
+    ...selection
   };
 }
 
@@ -308,6 +322,13 @@ export function createGcra({ rateMs, burst }) {
  */
 export function createRateLimitRegistry(limits = {}) {
   const limiters = new Map();
+  const matchingLimiter = (id, requestClass) => {
+    const entry = limiters.get(id);
+    return entry &&
+      (!entry.settings.requestClasses || entry.settings.requestClasses.includes(normalizeSchedulingClass(requestClass)))
+      ? entry
+      : null;
+  };
   const listeners = new Set();
   let wakeTimer = null;
   let wakeAt = Infinity;
@@ -370,7 +391,8 @@ export function createRateLimitRegistry(limits = {}) {
         entry.settings.maxConcurrent === settings.maxConcurrent &&
         entry.settings.rateMs === settings.rateMs &&
         entry.settings.burst === settings.burst &&
-        entry.settings.queue === settings.queue
+        entry.settings.queue === settings.queue &&
+        JSON.stringify(entry.settings.requestClasses) === JSON.stringify(settings.requestClasses)
       ) {
         continue;
       }
@@ -402,8 +424,10 @@ export function createRateLimitRegistry(limits = {}) {
       clearTimeout(wakeTimer);
       listeners.clear();
     },
-    tryAcquireChain(ids) {
-      const entries = [...new Set(ids)].map((id) => [id, limiters.get(id)]).filter(([, entry]) => entry);
+    tryAcquireChain(ids, { requestClass = 'standard' } = {}) {
+      const entries = [...new Set(ids)]
+        .map((id) => [id, matchingLimiter(id, requestClass)])
+        .filter(([, entry]) => entry);
       let waitMs = 0;
       for (const [id, entry] of entries) {
         const delay = entry.gcra?.nextAllowedInMs() ?? 0;
@@ -435,7 +459,7 @@ export function createRateLimitRegistry(limits = {}) {
         releases.push(release);
       }
       try {
-        consumeRateBudget({ limiter: (id) => limiters.get(id) }, ids);
+        consumeRateBudget({ limiter: (id) => matchingLimiter(id, requestClass) }, ids, { requestClass });
       } catch (error) {
         releases.forEach((fn) => fn());
         if (error.name !== 'RateBudgetExhaustedError') throw error;
@@ -450,8 +474,8 @@ export function createRateLimitRegistry(limits = {}) {
         notify();
       };
     },
-    limiter(id) {
-      return limiters.get(id) ?? null;
+    limiter(id, { requestClass } = {}) {
+      return requestClass === undefined ? (limiters.get(id) ?? null) : matchingLimiter(id, requestClass);
     },
     get size() {
       return limiters.size;
@@ -463,6 +487,7 @@ export function createRateLimitRegistry(limits = {}) {
         rateMs: entry.settings.rateMs ?? null,
         burst: entry.settings.burst ?? null,
         queue: entry.settings.queue === true,
+        requestClasses: entry.settings.requestClasses ?? [...REQUEST_CLASSES],
         active: entry.semaphore?.active ?? 0,
         queued: (entry.semaphore?.queued ?? 0) + entry.rateQueued,
         rateLimited: entry.gcra ? entry.gcra.nextAllowedInMs(now) > 0 : false,
@@ -476,13 +501,37 @@ export function createRateLimitRegistry(limits = {}) {
 export async function acquireRateLimitSlot(
   registry,
   id,
-  { signal = null, timeoutMs = MAX_WAIT_MS, rateBudget = true } = {}
+  { signal = null, timeoutMs = MAX_WAIT_MS, rateBudget = true, requestClass = 'standard' } = {}
 ) {
-  const entry = registry.limiter(id);
+  const entry = registry.limiter(id, { requestClass });
   if (!entry) return () => {};
   // Concurrency waits in a bounded FIFO queue, exactly like runtime slots.
   signal?.throwIfAborted?.();
-  const releaseSemaphore = entry.semaphore ? await entry.semaphore.acquire(signal, { timeoutMs }) : () => {};
+  let releaseSemaphore = () => {};
+  if (entry.semaphore) {
+    // A class-selection reload can exempt a request already in the legacy
+    // semaphore queue. Remove that waiter without interrupting admitted work.
+    const exemption = new AbortController();
+    const unsubscribe = registry.subscribe?.(() => {
+      if (!registry.limiter(id, { requestClass })) exemption.abort();
+    });
+    try {
+      releaseSemaphore = await entry.semaphore.acquire(
+        signal ? AbortSignal.any([signal, exemption.signal]) : exemption.signal,
+        { timeoutMs }
+      );
+    } catch (error) {
+      signal?.throwIfAborted?.();
+      if (!exemption.signal.aborted) throw error;
+      return () => {};
+    } finally {
+      unsubscribe?.();
+    }
+    if (!registry.limiter(id, { requestClass })) {
+      releaseSemaphore();
+      return () => {};
+    }
+  }
   try {
     signal?.throwIfAborted?.();
     if (rateBudget && entry.gcra) {
@@ -516,10 +565,10 @@ export async function acquireRateLimitSlot(
  * atomic: if any limiter rejects, every token already taken is refunded, so a
  * request that never reached a model costs no budget on the scopes it passed.
  */
-export function consumeRateBudget(registry, ids) {
+export function consumeRateBudget(registry, ids, { requestClass = 'standard' } = {}) {
   const consumed = [];
   for (const id of ids) {
-    const gcra = registry.limiter(id)?.gcra;
+    const gcra = registry.limiter(id, { requestClass })?.gcra;
     if (!gcra) continue;
     const decision = gcra.tryAcquire();
     if (decision.allowed) {
@@ -546,15 +595,16 @@ export function consumeRateBudget(registry, ids) {
 export async function awaitRateBudget(
   registry,
   ids,
-  { signal = null, started = Date.now(), capacity = RATE_LIMIT_QUEUE_CAPACITY } = {}
+  { signal = null, started = Date.now(), capacity = RATE_LIMIT_QUEUE_CAPACITY, requestClass = 'standard' } = {}
 ) {
   const maxQueued = Number.isSafeInteger(capacity) && capacity >= 0 ? capacity : RATE_LIMIT_QUEUE_CAPACITY;
   for (;;) {
     signal?.throwIfAborted?.();
     try {
-      return consumeRateBudget(registry, ids);
+      return consumeRateBudget(registry, ids, { requestClass });
     } catch (error) {
-      const entry = error?.name === 'RateBudgetExhaustedError' ? registry.limiter(error.limiterId) : null;
+      const entry =
+        error?.name === 'RateBudgetExhaustedError' ? registry.limiter(error.limiterId, { requestClass }) : null;
       if (!entry?.settings.queue || Date.now() + error.retryAfterMs - started > MAX_WAIT_MS) throw error;
       // Reserve before yielding: a concurrent caller can have passed the
       // server's precheck while this caller was acquiring semaphore slots.
@@ -563,24 +613,33 @@ export async function awaitRateBudget(
       }
       entry.rateQueued += 1;
       try {
-        await delay(error.retryAfterMs, signal);
+        await delay(error.retryAfterMs, signal, registry);
       } finally {
         entry.rateQueued -= 1;
       }
     }
   }
 }
 
-function delay(ms, signal) {
+function delay(ms, signal, registry) {
   return new Promise((resolve, reject) => {
-    const onAbort = () => {
+    let timer;
+    let unsubscribe;
+    const cleanup = () => {
       clearTimeout(timer);
+      unsubscribe?.();
+      signal?.removeEventListener?.('abort', onAbort);
+    };
+    const onAbort = () => {
+      cleanup();
       reject(signal.reason ?? abortError());
     };
-    const timer = setTimeout(() => {
-      signal?.removeEventListener?.('abort', onAbort);
+    const finish = () => {
+      cleanup();
       resolve();
-    }, ms);
+    };
+    timer = setTimeout(finish, ms);
+    unsubscribe = registry?.subscribe?.(finish);
     signal?.addEventListener?.('abort', onAbort, { once: true });
   });
 }

src/request-class.mjs

@@ -0,0 +1,5 @@
+export const REQUEST_CLASSES = ['standard', 'interactive', 'foreground'];
+
+export function normalizeSchedulingClass(value) {
+  return typeof value === 'string' && REQUEST_CLASSES.includes(value) ? value : 'standard';
+}

src/request-scheduler.mjs

@@ -1,4 +1,7 @@
 import { currentNodeId } from './cluster.mjs';
+import { REQUEST_CLASSES, normalizeSchedulingClass } from './request-class.mjs';
+
+export { REQUEST_CLASSES, normalizeSchedulingClass } from './request-class.mjs';
 
 // Priority scheduling for connected gateway inference requests.
 //
@@ -9,12 +12,6 @@ import { currentNodeId } from './cluster.mjs';
 // back without a backlog burst. Nothing here pauses or resumes a GPU: an
 // accepted request keeps its stream and simply waits its turn.
 
-export const REQUEST_CLASSES = ['standard', 'interactive', 'foreground'];
-
-export function normalizeSchedulingClass(value) {
-  return typeof value === 'string' && REQUEST_CLASSES.includes(value) ? value : 'standard';
-}
-
 function classRank(requestClass) {
   return REQUEST_CLASSES.indexOf(normalizeSchedulingClass(requestClass));
 }
@@ -271,14 +268,16 @@ export class RequestScheduler {
     return standard.indexOf(lease) < this.capacity(state);
   }
 
-  higherDemandForScopes(lease, scopeIds) {
+  higherDemandForScopes(lease, scopeIds, appliesToClass = () => true) {
     if (this.mode !== 'adaptive' || !scopeIds.length) return false;
     const scopes = new Set(scopeIds);
     const rank = classRank(lease.requestClass);
     for (const state of this.groups.values()) {
       if (
         [...state.leases, ...state.waiters].some(
-          (entry) => classRank(entry.requestClass) > rank && entry.scopeIds?.some((id) => scopes.has(id))
+          (entry) =>
+            classRank(entry.requestClass) > rank &&
+            entry.scopeIds?.some((id) => scopes.has(id) && appliesToClass(id, entry.requestClass))
         )
       )
         return true;