Setting the file. One moment.
Throttle · Vercel Optimize · vercel-labs/agent-skills · Skills Docs
Back to the top of the page — line 224
This file
Number 7.149
Position 149 of 155
Type JavaScript
Size 9 KB
Lines 273 lib/ throttle.mjs
JavaScript · 273 lines · 9 KB
10
const
DEFAULT_RATE_LIMIT
=
80
;
11 const DEFAULT_RATE_WINDOW_MS = 60_000 ;
12 const DAILY_OBSERVABILITY_LIMIT_RE = /daily . * observability . * query limit/ i ;
13
14 let dailyQuotaBlock = null ;
15
16 export function resolveConcurrency () {
17 return parsePositiveIntEnv ( 'VERCEL_OPTIMIZE_METRIC_CONCURRENCY' , DEFAULT_CONCURRENCY );
18 }
19
20 // Format: VERCEL_OPTIMIZE_METRIC_RATE=N or N/60s.
21 export function resolveRateLimit () {
22 const env = process.env. VERCEL_OPTIMIZE_METRIC_RATE ;
23 if (env == null || env === '' ) return { maxCalls: DEFAULT_RATE_LIMIT , windowMs: DEFAULT_RATE_WINDOW_MS };
24 const m = String (env). trim (). match ( / ^ ( \d + )(?: \/ ( \d + )( [sm] ) ? ) ?$ / );
25 if ( ! m) return { maxCalls: DEFAULT_RATE_LIMIT , windowMs: DEFAULT_RATE_WINDOW_MS };
26 const maxCalls = Number (m[ 1 ]);
27 if ( ! Number. isInteger (maxCalls) || maxCalls < 1 ) {
28 return { maxCalls: DEFAULT_RATE_LIMIT , windowMs: DEFAULT_RATE_WINDOW_MS };
29 }
30 if ( ! m[ 2 ]) return { maxCalls, windowMs: DEFAULT_RATE_WINDOW_MS };
31 const unit = m[ 3 ] === 'm' ? 60_000 : 1_000 ;
32 const windowMs = Number (m[ 2 ]) * unit;
33 return { maxCalls, windowMs };
34 }
35
36 function parsePositiveIntEnv ( name , defaultValue ) {
37 const env = process.env[name];
38 if (env == null || env === '' ) return defaultValue;
39 const n = Number (env);
40 if ( ! Number. isFinite (n) || n < 1 || ! Number. isInteger (n)) return defaultValue;
41 return n;
42 }
43
44 // FIFO semaphore. Caller MUST call returned release() exactly once.
45 export class SemaphoreAbortError extends Error {
46 constructor ( result ) {
47 super ( 'Semaphore acquire aborted' );
48 this .name = 'SemaphoreAbortError' ;
49 this .result = result;
50 }
51 }
52
53 export class Semaphore {
54 constructor ( max ) {
55 if ( ! Number. isInteger (max) || max < 1 ) {
56 throw new Error ( `Semaphore: max must be a positive integer (got ${ max })` );
57 }
58 this .max = max;
59 this .inFlight = 0 ;
60 this .waiters = [];
61 }
62
63 async acquire ( opts = {}) {
64 const abortIf = opts.abortIf;
65 const preAbort = abortIf ?.();
66 if (preAbort) throw new SemaphoreAbortError (preAbort);
67 if ( this .inFlight < this .max) {
68 this .inFlight ++ ;
69 return () => this . release ();
70 }
71 await new Promise (( resolve ) => this .waiters. push (resolve));
72 const postAbort = abortIf ?.();
73 if (postAbort) {
74 this . wakeNext ();
75 throw new SemaphoreAbortError (postAbort);
76 }
77 this .inFlight ++ ;
78 return () => this . release ();
79 }
80
81 release () {
82 this .inFlight -- ;
83 this . wakeNext ();
84 }
85
86 wakeNext () {
87 const next = this .waiters. shift ();
88 if (next) next ();
89 }
90
91 async run ( fn , opts = {}) {
92 const release = await this . acquire (opts);
93 try {
94 return await fn ();
95 } finally {
96 release ();
97 }
98 }
99 }
100
101 // Load-bearing — semaphore alone is insufficient (8 concurrent × ~1s queries = 480/min, well above the 100/min cap).
102 export class SlidingWindowRateLimiter {
103 constructor ( maxCalls , windowMs , opts = {}) {
104 if ( ! Number. isInteger (maxCalls) || maxCalls < 1 ) {
105 throw new Error ( `SlidingWindowRateLimiter: maxCalls must be >=1 (got ${ maxCalls })` );
106 }
107 if ( ! Number. isFinite (windowMs) || windowMs < 1 ) {
108 throw new Error ( `SlidingWindowRateLimiter: windowMs must be >0 (got ${ windowMs })` );
109 }
110 this .maxCalls = maxCalls;
111 this .windowMs = windowMs;
112 this .timestamps = []; // ascending order
113 this .now = opts.now ?? (() => Date. now ());
114 this .sleep = opts.sleep ?? defaultSleep;
115 }
116
117 async acquire () {
118 while ( true ) {
119 this . prune ();
120 if ( this .timestamps. length < this .maxCalls) {
121 this .timestamps. push ( this . now ());
122 return ;
123 }
124 // Small buffer avoids racing the window boundary.
125 const oldestExpiresAt = this .timestamps[ 0 ] + this .windowMs;
126 const sleepMs = Math. max ( 50 , oldestExpiresAt - this . now () + 100 );
127 await this . sleep (sleepMs);
128 }
129 }
130
131 prune () {
132 const cutoff = this . now () - this .windowMs;
133 while ( this .timestamps. length > 0 && this .timestamps[ 0 ] < cutoff) {
134 this .timestamps. shift ();
135 }
136 }
137 }
138
139 // Composes Semaphore + RateLimiter: bounds both burst (8 concurrent) and sustained throughput (80/60s).
140 let metricThrottleSingleton = null ;
141 export function getMetricThrottle () {
142 if ( ! metricThrottleSingleton) {
143 const semaphore = new Semaphore ( resolveConcurrency ());
144 const { maxCalls , windowMs } = resolveRateLimit ();
145 const rateLimiter = new SlidingWindowRateLimiter (maxCalls, windowMs);
146 metricThrottleSingleton = {
147 semaphore,
148 rateLimiter,
149 maxCalls,
150 windowMs,
151 async run ( fn ) {
152 const cached = getDailyQuotaBlock ();
153 if (cached) return dailyQuotaResult (cached);
154 let release;
155 try {
156 release = await semaphore. acquire ({ abortIf : () => {
157 const block = getDailyQuotaBlock ();
158 return block ? dailyQuotaResult (block) : null ;
159 } });
160 } catch (err) {
161 if (err instanceof SemaphoreAbortError ) return err.result;
162 throw err;
163 }
164 try {
165 const afterAcquire = getDailyQuotaBlock ();
166 if (afterAcquire) return dailyQuotaResult (afterAcquire);
167 await rateLimiter. acquire ();
168 const result = await fn ();
169 if ( isDailyQuotaExceeded (result)) {
170 const block = setDailyQuotaBlocked (result);
171 return dailyQuotaResult (block, result);
172 }
173 return result;
174 } finally {
175 release ();
176 }
177 },
178 };
179 }
180 return metricThrottleSingleton;
181 }
182
183 // Back-compat alias — returns the throttle object (compatible `.run(fn)` shape).
184 export const getMetricSemaphore = getMetricThrottle;
185
186 export function _resetMetricSemaphoreForTests () {
187 metricThrottleSingleton = null ;
188 dailyQuotaBlock = null ;
189 }
190
191 export async function retryOnRateLimit ( fn , opts = {}) {
192 const maxRetries = opts.maxRetries ?? DEFAULT_MAX_RETRIES ;
193 const baseBackoffMs = opts.baseBackoffMs ?? BASE_BACKOFF_MS ;
194 const jitterMs = opts.jitterMs ?? JITTER_MS ;
195 const sleep = opts.sleep ?? defaultSleep;
196 const onRetry = opts.onRetry;
197
198 let attempt = 0 ;
199 while ( true ) {
200 const result = await fn ();
201 if ( ! isRateLimited (result) || attempt >= maxRetries) return result;
202 attempt ++ ;
203 // attempt 1 = 1x, 2 = 1.5x, 3 = 2x of base.
204 const factor = 1 + (attempt - 1 ) * 0.5 ;
205 const jitter = jitterMs > 0 ? Math. random () * jitterMs : 0 ;
206 const delay = Math. round (baseBackoffMs * factor + jitter);
207 if (onRetry) onRetry (attempt, delay, result);
208 await sleep (delay);
209 }
210 }
211
212 // Variants: code='RATE_LIMITED' (canonical), 'rate_limited', or 'EXIT_1' + stderr match.
213 export function isRateLimited ( result ) {
214 if ( ! result || result.ok !== false ) return false ;
215 const code = String (result.code ?? '' ). toLowerCase ();
216 if (code === 'rate_limited' || code === '429' ) return true ;
217 const stderr = String (result.stderr ?? '' ). toLowerCase ();
218 if (stderr. includes ( 'rate limit' ) || stderr. includes ( 'rate_limited' ) || stderr. includes ( 'too many requests' )) {
219 return true ;
220 }
221 return false ;
222 }
223
224 export function isDailyQuotaExceeded ( result ) {
225 if ( ! result || result.ok !== false ) return false ;
226 const code = String (result.code ?? '' );
227 if (code. toUpperCase () === 'DAILY_QUOTA_EXCEEDED' ) return true ;
228 const haystack = [
229 result.message,
230 result.stderr,
231 result.stdout,
232 result.detail,
233 ]. filter (Boolean). join ( ' \n ' );
234 return DAILY_OBSERVABILITY_LIMIT_RE . test (haystack);
235 }
236
237 export function setDailyQuotaBlocked ( result , nowMs = Date. now ()) {
238 dailyQuotaBlock = {
239 untilMs: utcMidnightAfter (nowMs),
240 originalCode: result?.code ?? null ,
241 message: result?.message || result?.stderr || 'Daily Observability query limit reached.' ,
242 };
243 return dailyQuotaBlock;
244 }
245
246 export function getDailyQuotaBlock ( nowMs = Date. now ()) {
247 if ( ! dailyQuotaBlock) return null ;
248 if (dailyQuotaBlock.untilMs <= nowMs) {
249 dailyQuotaBlock = null ;
250 return null ;
251 }
252 return dailyQuotaBlock;
253 }
254
255 export function utcMidnightAfter ( nowMs ) {
256 const d = new Date (nowMs);
257 return Date. UTC (d. getUTCFullYear (), d. getUTCMonth (), d. getUTCDate () + 1 );
258 }
259
260 function dailyQuotaResult ( block , sourceResult = null ) {
261 return {
262 ... (sourceResult && typeof sourceResult === 'object' ? sourceResult : {}),
263 ok: false ,
264 code: 'DAILY_QUOTA_EXCEEDED' ,
265 message: block.message,
266 cachedUntil: new Date (block.untilMs). toISOString (),
267 originalCode: sourceResult?.originalCode ?? sourceResult?.code ?? block.originalCode ?? undefined ,
268 };
269 }
270
271 function defaultSleep ( ms ) {
272 return new Promise (( resolve ) => setTimeout (resolve, ms));
273 }