pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-09-15 19:32:20 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(quota): bound routing caches with shared boundedMap factory (#13280)

Browse Source 
Seven routing/quota caches (quality states, account buckets, quota-fetcher/saturation/header caches, learned rate limits) sit behind a shared bounded map with LRU/TTL eviction instead of growing without bound. The learned-limits cap of 200 that the tip declared was never enforced.

Maintainer rework before merge (kept the idea, no default behavior change):
- Eviction logging goes through the project logger, aggregated (first eviction, then one summary line per minute per map) instead of a `console.warn` per eviction.
- `refetch-lazy` and `hard-expire` behaved identically and are collapsed into `ttl`; protected entries (saturated account buckets, evaluator quality scores) are never evicted; caps raised to 2048–4096 so normal deployments never evict, with tests showing 300 learned limits and 600 cached entries all kept.

Validated first on the combined board of all 38 PRs of this batch (10 merged as-is, 28 after the maintainer rework) on top of release/v3.8.51 c0f92ec: typecheck:core, check:open-sse-typecheck and check:dashboard-typecheck clean; ESLint clean on every changed file; file-size (rebaselined for the combined growth), complexity, cognitive-complexity, changelog-integrity, docs-counts, docs-sync, migration-numbering and i18n new-key gates green; 735 focused node:test cases with the only batch-caused failure (a flag-count assertion) fixed. Then re-validated alone on the fresh release tip right before this merge: ESLint on the changed files, typecheck:core, check:open-sse-typecheck, the file-size/complexity/changelog gates and this PR's own tests.

Thanks @maxmad64bis!
This commit is contained in:
Dizzle
2026-09-15 18:08:21 +02:00
committed by GitHub
parent 215ac43a70
commit 7cabac4985
9 changed files with 589 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog.d/fixes/13280-bounded-routing-caches.md Normal file
View File

@@ -0,0 +1 @@
- **fix(quota):** in-process routing and quota caches (quality tracker, saturation and rate-limit header caches, quota-fetcher cache, learned rate limits, account buckets) are now size-bounded through one shared `boundedMap` — caps sit far above normal deployments, evictions are logged once per minute per cache instead of per entry, and state whose loss would change routing (live saturated quota buckets, evaluator quality scores) is never evicted ([#13280](https://github.com/diegosouzapw/OmniRoute/pull/13280)) — thanks @maxmad64bis

2
config/quality/eslint-suppressions.json
View File

@@ -585,7 +585,7 @@
},
"open-sse/services/rateLimitManager.ts": {
"@typescript-eslint/no-unused-vars": {
"count": 2
"count": 1
}
},
"open-sse/services/routing/index.ts": {

69
open-sse/services/genericQuotaFetcher.ts
View File

@@ -24,10 +24,8 @@ import {
type QuotaFetcher,
type QuotaInfo,
} from "./quotaPreflight.ts";
import {
getAntigravityQuotaFamily,
getQuotaFetchScope,
} from "./antigravityQuotaFamily.ts";
import { getAntigravityQuotaFamily, getQuotaFetchScope } from "./antigravityQuotaFamily.ts";
import { boundedMap } from "../../src/lib/quota/boundedMap.ts";
type UsageFetcher = (
connection: Parameters<typeof getUsageForProvider>[0],
@@ -77,29 +75,19 @@ export function __resetGenericQuotaFetcherForTests(): void {
pendingForceRefreshMiss.clear();
}
interface CacheEntry {
quota: QuotaInfo;
fetchedAt: number;
}
const cache = new Map<string, CacheEntry>();
// One entry per (provider, connection); 4096 keeps even very large account pools
// from ever evicting. An evicted entry only costs one extra upstream quota read.
const cache = boundedMap<QuotaInfo>("quota-fetcher-cache", 4096, "ttl", CACHE_TTL_MS);
function connectionKey(provider: string, connectionId: string): string {
return `${provider.trim()}::${connectionId.trim()}`;
}
function quotaCacheScope(
provider: string,
requestedModel?: string | null
): string {
function quotaCacheScope(provider: string, requestedModel?: string | null): string {
return getQuotaFetchScope(provider, requestedModel);
}
function cacheKey(
provider: string,
connectionId: string,
requestedModel?: string | null
): string {
function cacheKey(provider: string, connectionId: string, requestedModel?: string | null): string {
return `${connectionKey(provider, connectionId)}::${quotaCacheScope(provider, requestedModel)}`;
}
@@ -125,22 +113,14 @@ function markPendingForceRefreshMiss(key: string): void {
if (isPendingForceRefresh(key)) pendingForceRefreshMiss.set(key, Date.now());
}
function cachedQuotaIfFresh(
key: string,
forceRefresh: boolean,
now: number
): QuotaInfo | null {
function cachedQuotaIfFresh(key: string, forceRefresh: boolean, now: number): QuotaInfo | null {
if (forceRefresh) return null;
const cached = cache.get(key);
if (cached && now - cached.fetchedAt < CACHE_TTL_MS) return cached.quota;
const cached = cache.get(key, now);
if (cached !== undefined) return cached;
return null;
}
function isForceRefreshMissCooling(
key: string,
forceRefresh: boolean,
now: number
): boolean {
function isForceRefreshMissCooling(key: string, forceRefresh: boolean, now: number): boolean {
if (!forceRefresh) return false;
const missedAt = pendingForceRefreshMiss.get(key);
return missedAt !== undefined && now - missedAt < CACHE_TTL_MS;
@@ -150,18 +130,17 @@ function isForceRefreshMissCooling(
function isConcurrentForceRefresh(key: string, refreshStamp: number | undefined): boolean {
const currentStamp = pendingForceRefresh.get(key);
if (currentStamp === refreshStamp) return false;
return (
currentStamp !== undefined &&
Date.now() - currentStamp <= PENDING_FORCE_REFRESH_TTL_MS
);
return currentStamp !== undefined && Date.now() - currentStamp <= PENDING_FORCE_REFRESH_TTL_MS;
}
// 5min — same as Codex. Expiry is lazy on read (`isPendingForceRefresh`);
// this timer only reaps keys nobody fetches after the 5min TTL.
// 5min — same TTL as the original reap (CACHE_TTL_MS * 5). Expiry lazy on read
// (boundedMap ttl policy); this timer only keeps the sweep of
// pendingForceRefresh (5-min TTL, no systematic lazy read) + an opportunistic purge
// of stale cache entries along the way (get auto-purges).
const _cacheCleanup = setInterval(() => {
const now = Date.now();
for (const [key, entry] of cache) {
if (now - entry.fetchedAt > CACHE_TTL_MS * 5) cache.delete(key);
for (const key of cache.keys()) {
cache.get(key);
}
for (const key of pendingForceRefresh.keys()) {
dropExpiredPendingForceRefresh(key, now);
@@ -289,10 +268,7 @@ export function convertUsageToQuotaInfo(
const normalized = normalizeQuotaWindows(providerScopedWindows, context);
const scopedEntries = Object.values(providerScopedWindows);
const percentUsed = scopedEntries.reduce(
(worst, entry) => Math.max(worst, entry.percentUsed),
0
);
const percentUsed = scopedEntries.reduce((worst, entry) => Math.max(worst, entry.percentUsed), 0);
const resetAt =
scopedEntries.reduce<{ percentUsed: number; resetAt: string | null } | null>(
(worst, entry) => (!worst || entry.percentUsed > worst.percentUsed ? entry : worst),
@@ -322,10 +298,7 @@ function isAntigravityProvider(provider: string | null | undefined): boolean {
return provider === "antigravity" || provider === "agy";
}
function antigravityWeeklyWindowMatchesFamily(
key: string,
family: "gemini" | "claude"
): boolean {
function antigravityWeeklyWindowMatchesFamily(key: string, family: "gemini" | "claude"): boolean {
if (!key.endsWith("_weekly")) return false;
return family === "gemini" ? key === "gemini_weekly" : key === "claude_gpt_weekly";
}
@@ -456,7 +429,7 @@ export const fetchGenericQuota: QuotaFetcher = async (connectionId, connection)
const unscopedQuota = convertUsageToQuotaInfo(usage, { provider });
registerQuotaWindows(provider, Object.keys(unscopedQuota?.windows || quota.windows || {}));
cache.set(key, { quota, fetchedAt: Date.now() });
cache.set(key, quota);
return quota;
};

29
open-sse/services/rateLimitManager.ts
View File

@@ -39,6 +39,7 @@ import {
getExecutorTimeoutMs,
resolveConnectionTimeoutMs,
} from "../handlers/chatCore/upstreamTimeouts.ts";
import { boundedMap } from "../../src/lib/quota/boundedMap.ts";
interface LearnedLimitEntry {
provider: string;
@@ -90,8 +91,12 @@ const enabledConnections = new Set<string>();
const connectionRateLimitOverrides = new Map<string, Record<string, number>>();
// Store learned limits for persistence (debounced)
const learnedLimits: Record<string, LearnedLimitEntry> = {};
const MAX_LEARNED_LIMITS = 200;
// One learned entry per limiter key (provider:connection[:model]). The previous
// `MAX_LEARNED_LIMITS = 200` was declared but never enforced; enforcing 200 would
// start evicting (dropping persisted limits) on deployments with many
// connection×model limiters, so the enforced cap is set well above that.
export const MAX_LEARNED_LIMITS = 2048;
const learnedLimits = boundedMap<LearnedLimitEntry>("learned-limits", MAX_LEARNED_LIMITS, "lru");
const limiterLastUsed = new Map<string, number>();
let persistTimer: ReturnType<typeof setTimeout> | null = null;
const pendingAsyncOperations = new Set<Promise<unknown>>();
@@ -969,7 +974,7 @@ export function getAllRateLimitStatus() {
* Get all learned limits (for dashboard display).
*/
export function getLearnedLimits() {
return { ...learnedLimits };
return { ...Object.fromEntries(learnedLimits) };
}
// ─── Persistence ────────────────────────────────────────────────────────────
@@ -977,10 +982,8 @@ export function getLearnedLimits() {
async function persistLearnedLimitsNow() {
try {
const { updateSettings } = await import("@/lib/db/settings");
await updateSettings({ learnedRateLimits: JSON.stringify(learnedLimits) });
logRateLimit(
`💾 [RATE-LIMIT] Persisted learned limits for ${Object.keys(learnedLimits).length} provider(s)`
);
await updateSettings({ learnedRateLimits: JSON.stringify(Object.fromEntries(learnedLimits)) });
logRateLimit(`💾 [RATE-LIMIT] Persisted learned limits for ${learnedLimits.size} provider(s)`);
} catch (err) {
errorRateLimit("[RATE-LIMIT] Failed to persist learned limits:", err.message);
}
@@ -996,12 +999,12 @@ function recordLearnedLimit(
model: string | null = null
) {
const key = getLimiterKey(provider, connectionId, model);
learnedLimits[key] = {
learnedLimits.set(key, {
...limits,
provider,
connectionId,
lastUpdated: Date.now(),
};
});
// Debounce: save at most once per PERSIST_DEBOUNCE_MS
if (!persistTimer) {
@@ -1054,8 +1057,8 @@ export async function __resetRateLimitManagerForTests() {
limiterWatchdog.reset();
shutdownHandlersRegistered = false;
for (const key of Object.keys(learnedLimits)) {
delete learnedLimits[key];
for (const key of [...learnedLimits.keys()]) {
learnedLimits.delete(key);
}
if (pendingAsyncOperations.size > 0) {
@@ -1108,14 +1111,14 @@ async function loadPersistedLimits() {
const remaining = toNumber(data.remaining, 0);
const minTime = toNumber(data.minTime, 0);
learnedLimits[key] = {
learnedLimits.set(key, {
provider,
connectionId,
lastUpdated,
...(limit > 0 ? { limit } : {}),
...(remaining >= 0 ? { remaining } : {}),
...(minTime >= 0 ? { minTime } : {}),
};
});
// Apply to limiter if it exists and has rate limit enabled
if (connectionId && enabledConnections.has(connectionId)) {

18
open-sse/services/routing/quality.ts
View File

@@ -30,6 +30,7 @@
* Statistics are plain arithmetic (EWMA + small counters), O(1) per event, safe
* under the Node event loop's single thread — no lock-free/atomic trickery.
*/
import { boundedMap } from "../../../src/lib/quota/boundedMap.ts";
/** EWMA smoothing factor (alpha). Lower = slower adaptation. */
const OPERATIONAL_ALPHA = 0.2;
@@ -60,7 +61,18 @@ interface QualityState {
lastTs: number;
}
const states = new Map<string, QualityState>();
/**
* Cap on tracked (provider, model) pairs. Only pairs that actually carry traffic
* are tracked, so normal deployments stay far below it; past it the
* least-recently-used pair without an evaluator score is dropped (it restarts
* cold/neutral). Pairs holding a semantic score are never evicted — that score
* only comes from an evaluator run and cannot be re-learned from traffic.
*/
export const QUALITY_STATES_CAP = 4096;
const states = boundedMap<QualityState>("routing-quality", QUALITY_STATES_CAP, "lru", 0, {
shouldEvict: (s) => s.semantic === null,
});
function keyOf(provider: string, model: string): string {
return `${provider}/${model}`;
@@ -273,7 +285,9 @@ export function getQualityScore(provider: string, model: string): number {
/** Full snapshot of the tracker for explainability / dashboard. */
export function getQualitySnapshot(limit = 200): ProviderQuality[] {
const views: ProviderQuality[] = [];
for (const [key] of states) {
// Snapshot copy: LRU get refreshes recency (reinsertion), so iterating live + get()
// would loop forever. Snapshot behavior unchanged.
for (const [key] of [...states]) {
const slash = key.indexOf("/");
const provider = slash >= 0 ? key.slice(0, slash) : key;
const model = slash >= 0 ? key.slice(slash + 1) : key;

20
src/lib/quota/accountBuckets.ts
View File

@@ -16,6 +16,7 @@
*
* Part of: Quota Sharing Engine — Phase 3 (#3 multi-window buckets).
*/
import { boundedMap } from "./boundedMap";
// ---------------------------------------------------------------------------
// Types
@@ -62,8 +63,21 @@ export const SATURATION_THRESHOLD_PCT = 100;
// In-process store
// ---------------------------------------------------------------------------
/**
* Soft cap on stored buckets. Only SATURATED buckets are ever stored (a
* below-threshold observation deletes the entry), so evicting a live one would
* silently turn "saturated" into "eligible" — the fail-open this store exists to
* prevent. Over the cap only buckets whose reset instant already passed (stale
* saturation the next read would drop anyway) are evicted; live saturated
* buckets are never evicted and the store grows past the cap instead. 4096 ≈
* 1000+ connections × their 5h/7d/7d:<model> windows all saturated at once.
*/
export const ACCOUNT_BUCKETS_SOFT_CAP = 4096;
/** Key: `${connectionId}::${windowKey}`. */
const _buckets = new Map<string, BucketEntry>();
const _buckets = boundedMap<BucketEntry>("account-buckets", ACCOUNT_BUCKETS_SOFT_CAP, "lru", 0, {
shouldEvict: (entry, _key, nowMs) => entry.resetsAtMs > 0 && nowMs >= entry.resetsAtMs,
});
function storeKey(connectionId: string, windowKey: string): string {
return `${connectionId}::${windowKey}`;
@@ -104,7 +118,7 @@ export function isBucketSaturated(
): boolean {
if (!connectionId || !windowKey) return false; // fail-open
const key = storeKey(connectionId, windowKey);
const entry = _buckets.get(key);
const entry = _buckets.get(key, nowMs);
if (!entry) return false; // fail-open
// Lazy reset: the window rolled over → the saturation is stale.
@@ -156,7 +170,7 @@ export function recordUsage(
return;
}
_buckets.set(key, { saturated: true, resetsAtMs });
_buckets.set(key, { saturated: true, resetsAtMs }, nowMs);
}
/**

179
src/lib/quota/boundedMap.ts Normal file
View File

@@ -0,0 +1,179 @@
// src/lib/quota/boundedMap.ts — size-capped Map for in-process routing/quota caches.
import { createLogger } from "@/shared/utils/logger";
/**
* - `lru`: no expiry; over the cap the least-recently-USED entry goes first
* (`get` refreshes recency).
* - `ttl`: entries expire `ttlMs` after they were last `set` (expired reads return
* undefined and drop the entry); over the cap expired entries are swept first,
* then the oldest-WRITTEN entry goes (reads do not refresh anything).
*/
export type BoundedMapPolicy = "lru" | "ttl";
export interface BoundedMapLogger {
warn(meta: Record<string, unknown>, message: string): void;
}
export interface BoundedMapOptions<V> {
/**
* Return false to protect an entry from eviction. Protected entries are NEVER
* evicted: when every remaining entry is protected the map grows past its cap
* (and says so in the log) rather than dropping state whose loss would change
* routing — e.g. a saturated quota bucket (fail-open) or a semantic quality pin.
*/
shouldEvict?: (value: V, key: string, nowMs: number) => boolean;
/** Defaults to the project logger (`quota:bounded-map`). */
log?: BoundedMapLogger;
/** Minimum gap between two eviction log lines for one map. Default 60s. */
logIntervalMs?: number;
}
interface Entry<V> {
value: V;
ts: number;
}
export interface BoundedMap<V> {
get(key: string, nowMs?: number): V | undefined;
set(key: string, value: V, nowMs?: number): void;
delete(key: string): boolean;
clear(): void;
readonly size: number;
keys(): IterableIterator<string>;
[Symbol.iterator](): IterableIterator<[string, V]>;
/** Lifetime counters, for tests and diagnostics. */
stats(): { evictions: number; overflowInserts: number };
}
const DEFAULT_LOG_INTERVAL_MS = 60_000;
let defaultLogger: BoundedMapLogger | null = null;
function getDefaultLogger(): BoundedMapLogger {
defaultLogger ??= createLogger("quota:bounded-map");
return defaultLogger;
}
export function boundedMap<V>(
name: string,
limit: number,
policy: BoundedMapPolicy,
ttlMs = 0,
options: BoundedMapOptions<V> = {}
): BoundedMap<V> {
const inner = new Map<string, Entry<V>>();
const shouldEvict = options.shouldEvict ?? (() => true);
const logIntervalMs = options.logIntervalMs ?? DEFAULT_LOG_INTERVAL_MS;
const expires = policy === "ttl" && ttlMs > 0;
let evictions = 0;
let overflowInserts = 0;
// Aggregated logging: the first event logs at once, later ones are summed and
// reported at most once per logIntervalMs — a hot cache at its cap must not
// produce one log line per request.
let pendingEvictions = 0;
let pendingOverflows = 0;
let lastLogAt = Number.NEGATIVE_INFINITY;
function maybeLog(nowMs: number): void {
if (pendingEvictions === 0 && pendingOverflows === 0) return;
if (nowMs - lastLogAt < logIntervalMs) return;
lastLogAt = nowMs;
(options.log ?? getDefaultLogger()).warn(
{
map: name,
cap: limit,
size: inner.size,
evicted: pendingEvictions,
overflowInserts: pendingOverflows,
},
`[boundedMap:${name}] cap ${limit} reached: evicted ${pendingEvictions} entr${pendingEvictions === 1 ? "y" : "ies"}` +
(pendingOverflows > 0
? `, grew past the cap ${pendingOverflows}x (all entries protected)`
: "")
);
pendingEvictions = 0;
pendingOverflows = 0;
}
function isExpired(entry: Entry<V>, nowMs: number): boolean {
return expires && nowMs - entry.ts > ttlMs;
}
function sweepExpired(nowMs: number): void {
for (const [k, e] of inner) {
if (isExpired(e, nowMs)) inner.delete(k);
}
}
/** Map iteration order is recency (lru) or write order (ttl): the first evictable key wins. */
function findVictim(nowMs: number): string | undefined {
for (const [k, e] of inner) {
if (shouldEvict(e.value, k, nowMs)) return k;
}
return undefined;
}
function makeRoom(nowMs: number): void {
if (inner.size < limit) return;
if (expires) sweepExpired(nowMs);
while (inner.size >= limit) {
const victim = findVictim(nowMs);
if (victim === undefined) {
overflowInserts += 1;
pendingOverflows += 1;
break;
}
inner.delete(victim);
evictions += 1;
pendingEvictions += 1;
}
maybeLog(nowMs);
}
return {
get(key: string, nowMs: number = Date.now()): V | undefined {
const entry = inner.get(key);
if (!entry) return undefined;
if (isExpired(entry, nowMs)) {
inner.delete(key);
return undefined;
}
if (policy === "lru") {
inner.delete(key);
inner.set(key, entry);
}
return entry.value;
},
set(key: string, value: V, nowMs: number = Date.now()): void {
if (inner.has(key)) inner.delete(key);
else makeRoom(nowMs);
inner.set(key, { value, ts: nowMs });
},
delete(key: string): boolean {
return inner.delete(key);
},
clear(): void {
inner.clear();
},
get size(): number {
return inner.size;
},
keys(): IterableIterator<string> {
return inner.keys();
},
[Symbol.iterator](): IterableIterator<[string, V]> {
const nowMs = Date.now();
const it = inner.entries();
function* gen(): Generator<[string, V]> {
for (const [k, e] of it) {
if (isExpired(e, nowMs)) continue;
yield [k, e.value];
}
}
return gen();
},
stats() {
return { evictions, overflowInserts };
},
};
}

36
src/lib/quota/saturationSignals.ts
View File

@@ -24,6 +24,7 @@
*/
import { createLogger } from "@/shared/utils/logger";
import { boundedMap } from "./boundedMap";
import { updateAccountBuckets, type ClaudeUsageResult } from "./accountBuckets";
import type { QuotaUnit, QuotaWindow } from "./dimensions";
@@ -33,23 +34,20 @@ const log = createLogger("quota:saturation");
// Types
// ---------------------------------------------------------------------------
interface CacheEntry {
value: number; // 0..1
ts: number; // epoch ms
}
interface DimensionSpec {
unit: QuotaUnit;
window: QuotaWindow;
}
// ---------------------------------------------------------------------------
// In-memory cache (Map<cacheKey, CacheEntry>)
// In-memory cache (boundedMap<cacheKey, saturation 0..1>, 30s TTL). Caps are
// generous (one entry per connection/provider/dimension or provider/connection):
// an evicted entry only costs one extra read, and normal deployments never hit them.
// ---------------------------------------------------------------------------
const CACHE_TTL_MS = 30_000; // 30 seconds
const _cache = new Map<string, CacheEntry>();
const _cache = boundedMap<number>("saturation-cache", 4096, "ttl", CACHE_TTL_MS);
// Pending miss fetches, keyed like _cache. Concurrent getSaturation calls for
// the same key share the promise instead of firing one upstream read each.
@@ -79,9 +77,19 @@ interface TokenHeaderEntry {
ts: number;
}
const _rateLimitHeaders = new Map<string, RateLimitHeaderEntry>();
const _tokenHeaders = new Map<string, TokenHeaderEntry>();
const RL_HEADER_TTL_MS = 5 * 60 * 1000; // 5 minutes
const _rateLimitHeaders = boundedMap<RateLimitHeaderEntry>(
"saturation-rl-headers",
4096,
"ttl",
RL_HEADER_TTL_MS
);
const _tokenHeaders = boundedMap<TokenHeaderEntry>(
"saturation-token-headers",
4096,
"ttl",
RL_HEADER_TTL_MS
);
/** Test-only: clear the rate-limit + token header caches between asserts. */
export function _clearRateLimitHeaders(): void {
@@ -249,7 +257,7 @@ export function getTokenHeaderSaturation(
connectionId: string
): { saturation: number; resetAt: number | null } | null {
const entry = _tokenHeaders.get(`${provider}:${connectionId}`);
if (!entry || Date.now() - entry.ts > RL_HEADER_TTL_MS) return null;
if (!entry) return null;
if (!(entry.limit > 0)) return null;
const used = entry.limit - entry.remaining;
const saturation = Math.min(1, Math.max(0, used / entry.limit));
@@ -342,7 +350,7 @@ async function fetchBailianSaturation(connectionId: string, dim: DimensionSpec):
*/
function anthropicHeaderSaturation(connectionId: string): number {
const entry = _rateLimitHeaders.get(`anthropic:${connectionId}`);
if (!entry || Date.now() - entry.ts > RL_HEADER_TTL_MS) return 0;
if (!entry) return 0;
const used = entry.limit - entry.remaining;
return Math.min(1, Math.max(0, used / entry.limit));
@@ -538,8 +546,8 @@ export async function getSaturation(
): Promise<number> {
const key = cacheKey(connectionId, provider, dim);
const cached = _cache.get(key);
if (cached && Date.now() - cached.ts < CACHE_TTL_MS) {
return cached.value;
if (cached !== undefined) {
return cached;
}
const pending = _inflight.get(key);
@@ -569,7 +577,7 @@ export async function getSaturation(
);
value = 0;
}
_cache.set(key, { value, ts: Date.now() });
_cache.set(key, value);
return value;
})();
_inflight.set(key, task);

316
tests/unit/quota-bounded-map.test.ts Normal file
View File

@@ -0,0 +1,316 @@
import test from "node:test";
import assert from "node:assert/strict";
const { boundedMap } = await import("../../src/lib/quota/boundedMap.ts");
const core = await import("../../src/lib/db/core.ts");
type LogLine = { meta: Record<string, unknown>; message: string };
function captureLog() {
const lines: LogLine[] = [];
return {
lines,
log: {
warn: (meta: Record<string, unknown>, message: string) => lines.push({ meta, message }),
},
};
}
test.after(() => {
core.resetDbInstance();
});
// ── boundedMap primitives ────────────────────────────────────────────────────
test("lru: evicts the least-recently-used entry and get refreshes recency", () => {
const { log } = captureLog();
const m = boundedMap<number>("t", 3, "lru", 0, { log });
m.set("a", 1);
m.set("b", 2);
m.set("c", 3);
assert.equal(m.get("a"), 1); // a is now the most recent
m.set("d", 4); // evicts b
assert.equal(m.get("b"), undefined);
assert.equal(m.get("a"), 1);
assert.equal(m.size, 3);
assert.deepEqual(m.stats(), { evictions: 1, overflowInserts: 0 });
});
test("ttl: reads never refresh — the oldest-written entry is evicted", () => {
const { log } = captureLog();
const m = boundedMap<number>("t", 3, "ttl", 60_000, { log });
m.set("a", 1, 0);
m.set("b", 2, 1);
m.set("c", 3, 2);
assert.equal(m.get("a", 3), 1); // a read, but ttl ignores recency
m.set("d", 4, 4); // evicts a (oldest write), unlike lru which would evict b
assert.equal(m.get("a", 5), undefined);
assert.equal(m.get("b", 5), 2);
});
test("ttl: entries expire after ttlMs; lru entries never expire", () => {
const ttl = boundedMap<number>("t", 10, "ttl", 1000);
ttl.set("a", 1, 0);
assert.equal(ttl.get("a", 1000), 1);
assert.equal(ttl.get("a", 1001), undefined);
assert.equal(ttl.size, 0, "an expired read drops the entry");
const lru = boundedMap<number>("t", 10, "lru", 1000);
lru.set("a", 1, 0);
assert.equal(lru.get("a", 10_000_000), 1);
});
test("ttl: expired entries are swept before any fresh entry is evicted", () => {
const { log } = captureLog();
const m = boundedMap<number>("t", 3, "ttl", 100, { log });
m.set("old1", 1, 0);
m.set("fresh", 2, 150);
m.set("old2", 3, 0);
m.set("new", 4, 160); // old1 + old2 expired at 160 → swept, fresh survives
assert.equal(m.get("fresh", 170), 2);
assert.equal(m.get("new", 170), 4);
assert.equal(m.stats().evictions, 0, "a sweep of expired entries is not an eviction");
});
test("protected entries are never evicted: the map grows past the cap instead", () => {
const { log } = captureLog();
const m = boundedMap<{ pin: boolean }>("t", 2, "lru", 0, {
shouldEvict: (v) => !v.pin,
log,
});
m.set("pin1", { pin: true });
m.set("x", { pin: false });
m.set("y", { pin: false }); // evicts x, the only evictable entry
assert.equal(m.get("x"), undefined);
m.set("pin2", { pin: true }); // evicts y
m.set("pin3", { pin: true }); // nothing evictable → grows
assert.equal(m.size, 3);
for (const key of ["pin1", "pin2", "pin3"]) assert.deepEqual(m.get(key), { pin: true });
assert.deepEqual(m.stats(), { evictions: 2, overflowInserts: 1 });
});
test("eviction logging is aggregated and rate-limited, never one line per eviction", () => {
const { lines, log } = captureLog();
const m = boundedMap<number>("hot-cache", 10, "lru", 0, { log, logIntervalMs: 60_000 });
for (let i = 0; i < 10; i++) m.set(`seed-${i}`, i, 0);
for (let i = 0; i < 1000; i++) m.set(`k-${i}`, i, 1000 + i); // 1000 evictions in ~1s
assert.equal(lines.length, 1, "first eviction logs once, the rest are aggregated");
assert.equal(lines[0].meta.map, "hot-cache");
assert.equal(lines[0].meta.evicted, 1);
m.set("late", 1, 1000 + 61_000); // past the interval → one summary line
assert.equal(lines.length, 2);
assert.equal(
lines[1].meta.evicted,
1000,
"the summary carries every eviction since the last line"
);
assert.match(lines[1].message, /\[boundedMap:hot-cache\] cap 10 reached: evicted 1000 entries/);
});
test("the default logger is the project logger, not console.warn", () => {
const original = console.warn;
let consoleWarnings = 0;
console.warn = () => {
consoleWarnings += 1;
};
try {
const m = boundedMap<number>("console-check", 1, "lru");
m.set("a", 1);
m.set("b", 2);
m.set("c", 3);
assert.equal(m.stats().evictions, 2);
} finally {
console.warn = original;
}
assert.equal(consoleWarnings, 0);
});
test("keys() iteration tolerates delete during iteration", () => {
const m = boundedMap<number>("t", 10, "lru");
m.set("a", 1);
m.set("b", 2);
for (const key of m.keys()) {
if (key === "a") m.delete(key);
}
assert.deepEqual([...m.keys()], ["b"]);
});
// ── account buckets: never fail open ─────────────────────────────────────────
test("account buckets never evict a live saturated bucket, even past the soft cap", async () => {
const b = await import("../../src/lib/quota/accountBuckets.ts");
b._clearBucketsForTest();
const now = 1_800_000_000_000;
const future = new Date(now + 3_600_000).toISOString();
try {
const total = b.ACCOUNT_BUCKETS_SOFT_CAP + 25;
for (let i = 0; i < total; i++) b.recordUsage(`conn-live-${i}`, "5h", 100, future, now);
assert.equal(b._bucketCountForTest(), total, "no saturated bucket was dropped");
assert.equal(b.isBucketSaturated("conn-live-0", "5h", now + 1), true, "oldest still saturated");
assert.equal(b.isBucketSaturated(`conn-live-${total - 1}`, "5h", now + 1), true);
} finally {
b._clearBucketsForTest();
}
});
test("account buckets at the cap evict buckets whose reset already passed first", async () => {
const b = await import("../../src/lib/quota/accountBuckets.ts");
b._clearBucketsForTest();
const now = 1_800_000_000_000;
const soon = new Date(now + 1_000).toISOString();
const later = new Date(now + 3_600_000).toISOString();
try {
b.recordUsage("conn-stale", "5h", 100, soon, now); // resets 1s later
for (let i = 1; i < b.ACCOUNT_BUCKETS_SOFT_CAP; i++) {
b.recordUsage(`conn-keep-${i}`, "5h", 100, later, now);
}
assert.equal(b._bucketCountForTest(), b.ACCOUNT_BUCKETS_SOFT_CAP);
b.recordUsage("conn-new", "5h", 100, later, now + 5_000); // stale bucket is evictable now
assert.equal(b._bucketCountForTest(), b.ACCOUNT_BUCKETS_SOFT_CAP, "stale bucket made room");
assert.equal(b.isBucketSaturated("conn-keep-1", "5h", now + 5_001), true);
assert.equal(b.isBucketSaturated("conn-new", "5h", now + 5_001), true);
} finally {
b._clearBucketsForTest();
}
});
// ── quality tracker under real pressure ──────────────────────────────────────
test("quality: past the cap the LRU unscored pair is dropped, semantic pins survive", async () => {
const q = await import("../../open-sse/services/routing/quality.ts");
q.resetQualityTracker();
const event = (provider: string, model: string) => ({
provider,
model,
outcome: "success",
status: 200,
latencyMs: 100,
finishReason: "stop",
});
try {
q.recordQualityEvent(event("pinned", "model"));
q.setSemanticQuality("pinned", "model", 0.9, 1);
q.recordQualityEvent(event("first", "unscored"));
for (let i = 0; i < q.QUALITY_STATES_CAP + 50; i++) {
q.recordQualityEvent(event("bulk", `m-${i}`));
}
const snapshot = q.getQualitySnapshot(q.QUALITY_STATES_CAP * 2);
assert.equal(snapshot.length, q.QUALITY_STATES_CAP, "tracker stays at its cap");
const pinned = snapshot.find((v) => v.provider === "pinned");
assert.ok(pinned, "the semantic pin survived the pressure");
assert.equal(pinned.semantic, 0.9);
assert.equal(q.getProviderQuality("first", "unscored").samples, 0, "LRU unscored pair evicted");
assert.ok(q.getProviderQuality("bulk", `m-${q.QUALITY_STATES_CAP + 49}`).samples > 0);
} finally {
q.resetQualityTracker();
}
});
// ── learned rate limits ──────────────────────────────────────────────────────
const HEADERS = {
"x-ratelimit-limit-requests": "100",
"x-ratelimit-remaining-requests": "5",
"x-ratelimit-reset-requests": "30s",
};
test("learnedLimits: a deployment above the old unenforced 200 keeps every entry", async () => {
const rl = await import("../../open-sse/services/rateLimitManager.ts");
await rl.__resetRateLimitManagerForTests();
try {
for (let i = 0; i < 300; i++) {
rl.enableRateLimitProtection(`conn-many-${i}`);
rl.updateFromHeaders("openai", `conn-many-${i}`, HEADERS, 200);
}
assert.equal(Object.keys(rl.getLearnedLimits()).length, 300);
} finally {
await rl.__resetRateLimitManagerForTests();
}
});
test("learnedLimits: capped at MAX_LEARNED_LIMITS", async () => {
const rl = await import("../../open-sse/services/rateLimitManager.ts");
await rl.__resetRateLimitManagerForTests();
try {
for (let i = 0; i <= rl.MAX_LEARNED_LIMITS; i++) {
rl.enableRateLimitProtection(`conn-cap-${i}`);
rl.updateFromHeaders("openai", `conn-cap-${i}`, HEADERS, 200);
}
const learned = rl.getLearnedLimits();
assert.equal(Object.keys(learned).length, rl.MAX_LEARNED_LIMITS);
assert.equal(learned["openai:conn-cap-0"], undefined, "oldest entry evicted");
} finally {
await rl.__resetRateLimitManagerForTests();
}
});
test("learnedLimits persist/load round-trip", async () => {
const rl = await import("../../open-sse/services/rateLimitManager.ts");
const settings = await import("../../src/lib/db/settings.ts");
await rl.__resetRateLimitManagerForTests();
try {
rl.enableRateLimitProtection("conn-rt");
rl.updateFromHeaders("openai", "conn-rt", HEADERS, 200);
await rl.__flushLearnedLimitsForTests();
const raw = (await settings.getSettings())?.learnedRateLimits;
assert.equal(typeof raw, "string");
const parsed = JSON.parse(raw as string) as Record<string, { limit?: number }>;
assert.equal(parsed["openai:conn-rt"]?.limit, 100);
await rl.__resetRateLimitManagerForTests();
assert.deepEqual(rl.getLearnedLimits(), {});
await rl.initializeRateLimits();
assert.ok(rl.getLearnedLimits()["openai:conn-rt"], "load restores the persisted entry");
} finally {
await rl.__resetRateLimitManagerForTests();
}
});
// ── TTL caches keep their read-through behaviour ────────────────────────────
test("saturation cache: hits stay cached below the cap, the evicted key refetches past it", async () => {
const sat = await import("../../src/lib/quota/saturationSignals.ts");
sat._clearSaturationCache();
let calls = 0;
sat.__setGenericUsageFetcherForTests(async () => {
calls++;
return { percentUsed: 0.1 };
});
const dim = { unit: "tokens", window: "hourly" } as const;
try {
for (let i = 0; i < 600; i++) await sat.getSaturation(`conn-sat-${i}`, "some-provider", dim);
const warm = calls;
await sat.getSaturation("conn-sat-0", "some-provider", dim);
assert.equal(calls, warm, "600 entries (above the old 512 cap) are still cached");
for (let i = 600; i < 4097; i++) await sat.getSaturation(`conn-sat-${i}`, "some-provider", dim);
const before = calls;
// ttl policy: the read above did not refresh conn-sat-0, so as the oldest write
// it is the one entry evicted by the 4097th insert.
await sat.getSaturation("conn-sat-0", "some-provider", dim);
assert.ok(calls > before, `evicted key must refetch (calls ${before} -> ${calls})`);
} finally {
sat.__setGenericUsageFetcherForTests(null);
sat._clearSaturationCache();
}
});
test("quota-fetcher cache: entries above the old 512 cap stay cached", async () => {
const g = await import("../../open-sse/services/genericQuotaFetcher.ts");
g.__resetGenericQuotaFetcherForTests();
let calls = 0;
g.__setGenericUsageFetcherForTests(async () => {
calls++;
return { quotas: { session: { remainingPercentage: 50, resetAt: null } } };
});
try {
for (let i = 0; i < 600; i++) {
await g.fetchGenericQuota(`gqf-${i}`, { id: `gqf-${i}`, provider: "openai" });
}
const before = calls;
await g.fetchGenericQuota("gqf-0", { id: "gqf-0", provider: "openai" });
assert.equal(calls, before, "cache hit, no refetch");
} finally {
g.__setGenericUsageFetcherForTests(null);
g.__resetGenericQuotaFetcherForTests();
}
});