diff --git a/docs/guides/INTEGRATION-PATTERNS.md b/docs/guides/INTEGRATION-PATTERNS.md index eddba15fa..d1a0e102f 100644 --- a/docs/guides/INTEGRATION-PATTERNS.md +++ b/docs/guides/INTEGRATION-PATTERNS.md @@ -796,7 +796,17 @@ module.exports = { MyIntegration }; ### Rate Limiting -Always respect API rate limits: +Put what you know about a provider's limits in its API module: declare +`static rateLimit` (see the +[API module reference](../reference/api-module-definition-and-functions.md#rate-limits-static-ratelimit)). +The Requester then waits as the provider says. When the wait is too long to +sleep, it throws `RateLimitError` with `retryAt`. A queue handler checks +`delivery.isLastAttempt` first (see [Queue Handler Delivery](#queue-handler-delivery)) +and ends the run on the last attempt, then rethrows: the queue worker does not +halt a `RateLimitError`, and SQS redelivers the message. Do not catch it to +write a retry of your own. + +For pacing in your own loop, respect API rate limits: ```javascript async processBatchHandler({ processId, entityIds, entityType }) { diff --git a/docs/guides/LOGGING-IN-INTEGRATIONS.md b/docs/guides/LOGGING-IN-INTEGRATIONS.md index d7b3b2610..6b65771b5 100644 --- a/docs/guides/LOGGING-IN-INTEGRATIONS.md +++ b/docs/guides/LOGGING-IN-INTEGRATIONS.md @@ -189,7 +189,8 @@ try { error)` also works (the logger moves it to `error`), but `{ error }` is the standard form. - The logger writes the error as `{ type, message, code, status, stack, - cause }`, redacted. The `cause` chain is kept to 3 levels. Other own + cause }`, redacted, plus `retryAt` and `reason` for a `RateLimitError`. + The `cause` chain is kept to 3 levels. Other own properties of the error (for example axios `config`, `request`, `response`) are not written, because they often hold tokens. - A `FetchError` from an API module already has a safe message: diff --git a/docs/guides/LOGGING.md b/docs/guides/LOGGING.md index 53bfceedd..92367ec55 100644 --- a/docs/guides/LOGGING.md +++ b/docs/guides/LOGGING.md @@ -88,7 +88,7 @@ One record, shown on several lines (stdout has it on one line): | `entityId`, `credentialId` | string | `Module` bindings | in API module records | | `eventName` | string | call site: the `logger` name, then `.` | required for `frigg.*` at `WARN` and above, and for framework lifecycle `INFO` | | `trace_id`, `span_id`, `trace_flags` | string (hex) | the active OTel span | inside a span only | -| `error` | object | `{ type, message, code, status, stack, cause }`, sanitized | when a call site passes `error` | +| `error` | object | `{ type, message, code, status, stack, cause }`, sanitized; plus `retryAt` (ISO) and `reason` for an error with `isRateLimited` and a valid `retryAt` | when a call site passes `error` | | `droppedKeys` | string[] | logger | when the logger removed a field (see Precedence) | Call-site fields go next to these, at the top level. @@ -162,6 +162,14 @@ the app deploys defines the schema. layers throw with `cause` and do not log. `createHandler`, the express error middleware, `Worker.run` and the DLQ processor are the boundaries. +The `Requester` writes one `TRACE` record, `module..rate_limited`, for +each wait that a provider or a module policy set. It carries `statusCode`, +`waitMs`, `retryAt`, `reason`, `hintSource`, `attempt` and `waitedMs`. The +plain backoff ladder writes none. The `RateLimitError` it throws gets no record +of its own: the boundary logs it, and its `error` field carries `retryAt` and +`reason`. A `classify()` that throws or returns a Promise writes one `WARN`, +`module..rate_limit_classify_failed`. + `FRIGG_LOG_LEVEL` sets the minimum level. The logger drops a record below it before it reads any field. diff --git a/docs/reference/api-module-definition-and-functions.md b/docs/reference/api-module-definition-and-functions.md index e209be457..23712f64d 100644 --- a/docs/reference/api-module-definition-and-functions.md +++ b/docs/reference/api-module-definition-and-functions.md @@ -219,6 +219,218 @@ encryption: { credentialFields: ['signing_key', 'webhook_secret'] }, apiPropertiesToPersist: { credential: ['signing_key', 'webhook_secret'] } ``` +## Rate Limits (`static rateLimit`) + +An API module can tell the Requester how its provider limits calls, and how to +read a throttled response. Declare a static `rateLimit` on the API class. Every +key is optional. A module that declares nothing keeps the fixed backoff ladder +for a 429 (1, 3, 10, 30, 60 and 180 s), and reads the time from a 429 with the +default `parsers`: `Retry-After`, the reset headers and the IETF `RateLimit` +headers. The JSON schema still has an older `config.rateLimit: { requests, period }` +that the runtime does not read: only `static rateLimit` on the API class counts. + +| Key | Meaning | +|---|---| +| `scope` | The key a limit counts against: `'entity'` (default), `'credential'`, `'app'`, or a synchronous function `(requester) => key` that returns a string or a number. The Requester puts it on `RateLimitError.scopeKey`, which is `undefined` when the requester has no id for the scope or the function returns another type. | +| `minRetryAfterMs` | A wait never shorter than this. Use it when the docs say "wait 60 seconds" and the response says nothing. | +| `parsers` | The header parsers to read, in order: `retryAfter`, `resetHeaders`, `ietf`. Default: all three. A name that is not built in throws a `TypeError` when the API class is constructed. | +| `classify` | Recognises a limit that is not a plain 429, or names its reason. See below. | +| `windows`, `maxConcurrency` | The provider's limits, as documentation. A window named like the `reason` (`{ name: 'burst', perMs: 10000 }`) sets the wait when no header does, unless `minRetryAfterMs` is longer. Pacing uses them later. | +| `userHints` | Links per `reason` that a UI can show, for example `{ daily: { links: [{ label, url }] } }`. | + +### Built-in parsers + +| Parser | Reads | +|---|---| +| `retryAfter` | `Retry-After`: delta seconds, an HTTP-date or an ISO timestamp | +| `resetHeaders` | `X-RateLimit-Reset-After` (delta seconds), else `X-RateLimit-Reset`, `RateLimit-Reset` or `X-Rate-Limit-Reset`: epoch milliseconds, epoch seconds or delta seconds (told apart by size), or a date. `remaining` comes from `X-RateLimit-Remaining`, `RateLimit-Remaining` or `X-Rate-Limit-Remaining` | +| `ietf` | The IETF `RateLimit` field, as `limit=, remaining=, reset=` or as structured items (`"name";r=0;t=12`). In a list, the item with the fewest remaining calls wins, then the longest wait. `RateLimit-Policy` names the policy | + +A parser reads the headers from a `Headers` object, a `Map`, an entries array or +a plain object, and matches the names in any case. A time in the past waits +0 ms. A wait that is negative, not a number or longer than 366 days is ignored. +The package exports the parsers as `parseRetryAfter(value, { now })`, +`parseResetHeaders(headers, { now })` and `parseIetfRateLimit(headers, { now })`. +Each returns a hint or `null`. + +### What `classify` returns + +`classify({ status, headers, body })` runs for a 429, and for any other 4xx or +5xx except 401. `body` is the response text parsed as JSON, whatever the media +type, or `undefined` when the text is not JSON (an HTML error page, for +example). It returns `null` when the response is not a limit, or a hint: + +| Field | Meaning | +|---|---| +| `reason` | `'burst'`, `'daily'`, `'monthly'`, `'concurrency'` or `'unknown'` | +| `waitMs` or `retryAt` | When to call again. Leave both out to take the time from the parsers, then from the policy. Step 2 of [How the Requester waits](#how-the-requester-waits) says what happens when neither gives one. | +| `policy` | The provider's own name for the limit, for logs and the UI | +| `remaining` | Calls left in the window, when the response says | +| `source` | `'header'`, `'body'` (default) or `'static'` | + +A status other than 429 is a limit only when `classify` returns a hint for it. +A result counts only when it has a known `reason`, a finite `waitMs` or a valid +`retryAt`; anything else, `{}` for example, counts as `null`. `classify` must +be synchronous. A `classify` that throws or returns a Promise is logged +(`rate_limit_classify_failed`) and counts as `null`. + +```javascript +class Api extends OAuth2Requester { + static rateLimit = { + scope: 'entity', + windows: [{ name: 'burst', limit: 100, perMs: 10_000 }], + parsers: ['retryAfter', 'resetHeaders'], + classify({ status, body }) { + if (status !== 429) return null; + if (body?.policyName === 'DAILY') return { reason: 'daily', waitMs: 3_600_000 }; + return { reason: 'burst', policy: body?.policyName }; + }, + userHints: { + daily: { links: [{ label: 'API usage limits', url: 'https://example.com/limits' }] }, + }, + }; +} +``` + +```javascript +class Api extends ApiKeyRequester { + static rateLimit = { minRetryAfterMs: 60_000 }; +} +``` + +### How the Requester waits + +1. It looks for a hint in this order: `classify()`, the parsers, the policy + (the larger of `minRetryAfterMs` and the `perMs` of the window named by the + reason). +2. With no time, a 429 keeps the fixed ladder: the same calls, the same delays, + then a plain `FetchError`, as before. A response with another status that + `classify` recognises but that gets no time throws `RateLimitError` at once, + after one call, with `retryAt` and `waitMs` `undefined`: the time is unknown, + and a ladder of seconds cannot outwait a daily limit. A 5xx that `classify` + does not recognise keeps the 5xx ladder. +3. With a hint, the wait is the largest of the hint, `minRetryAfterMs` and 1 s, + plus at most 10 % jitter. The Requester sleeps when the wait fits the time + this request may still sleep, in total, and the time left in the Lambda + invocation less one request timeout. That reserve is never more than half + the time left, so a 29 s HTTP invocation with the default 60 s request + timeout can still sleep 14.5 s. The caller owns the total, not the module: + 10 s in an HTTP handler, 60 s in an integration queue worker, and 300 s + outside an invocation (tests, scripts, `frigg start`). See + [The invocation deadline](#the-invocation-deadline). A request makes at most + 10 hinted retries. They do not use the retries of the ladder, so a 5xx after + a hinted wait still gets the whole 5xx ladder. +4. A wait that does not fit throws `RateLimitError`. It extends `FetchError`, so + `statusCode` stays, and adds `isRateLimited`, `retryAt`, `waitMs`, `reason`, + `policy`, `source`, `module` and `scopeKey`. + +A Requester built with `backOff: []` never sleeps: a hinted wait throws +`RateLimitError` at once, and a 429 with no hint throws a plain `FetchError`. + +The parsers, `Retry-After` among them, read the headers of a 429 and of a +response that `classify` recognises, and of no other response: a 503 with +`Retry-After` keeps the 5xx ladder unless `classify` recognises it. The queue +worker halts a `RateLimitError` with a 4xx status other than 408 and 429, like +any other 4xx: it discards the message. + +### A client that is not the Requester + +A module that drives another client (for example jsforce) calls +`classifyRateLimit` around its own calls and throws the error itself: + +```javascript +const { classifyRateLimit, RateLimitError } = require('@friggframework/core'); + +async withLimits(call) { + try { + return await call(); + } catch (err) { + const hint = classifyRateLimit(Api.rateLimit, { + status: err.statusCode, + headers: {}, + body: { errorCode: err.errorCode }, + }); + if (!hint) throw err; + throw new RateLimitError({ hint, module: this.delegate?.name, cause: err }); + } +} +``` + +`classifyRateLimit(policy, { status, headers, body }, { now, onClassifyError })` +resolves in the order the Requester uses and returns a hint or `null`. A hint is +`{ retryAt, waitMs, reason, source, policy, remaining }`: `retryAt` is a `Date`, +`waitMs` counts from the call, and `policy` and `remaining` are there only when +known. An error that `classify` throws propagates unless you pass +`onClassifyError`. + +`new RateLimitError({ hint, waitMs, module, scopeKey, ...fetchErrorArgs })` +takes the `FetchError` arguments (`resource`, `init`, `response`, `cause`, +`responseBody`). It sets `retryAt` to now plus `waitMs`, or to `hint.retryAt` +when `waitMs` is left out. With no `response`, `statusCode` comes from +`cause.statusCode`. + +### A limit in a 2xx body + +The Requester calls `classify` only for a 4xx or 5xx: a generic retry of a +"throttled 200" could send a POST twice. A provider that answers 200 with a +limit error (a GraphQL `THROTTLED` error, for example) needs the module to +wrap its own call, classify the body and throw: + +```javascript +const { classifyRateLimit, RateLimitError } = require('@friggframework/core'); + +class Api extends OAuth2Requester { + static rateLimit = { + classify({ body }) { + const throttled = body?.errors?.some( + (error) => error.extensions?.code === 'THROTTLED' + ); + return throttled ? { reason: 'burst', waitMs: 2_000 } : null; + }, + }; + + async query(document, variables) { + const url = `${this.baseUrl}/graphql`; + const response = await this._post({ + url, + body: { query: document, variables }, + returnFullRes: true, + }); + const body = await response.json(); + if (!body.errors?.length) return body.data; + + const cause = new Error(body.errors[0].message); + const hint = classifyRateLimit(Api.rateLimit, { + status: response.status, + headers: response.headers, + body, + }); + if (!hint) throw cause; + throw new RateLimitError({ hint, module: this.delegate?.name, cause, resource: url }); + } +} +``` + +### The invocation deadline + +`runInvocationScope` stores two limits for every Lambda invocation: the end of +the invocation, from `context.getRemainingTimeInMillis()`, and the most one +Requester call may sleep in process, in total. `createHandler` takes the second +as the option `maxInProcessWaitMs`. + +| Entry point | `maxInProcessWaitMs` | +|---|---| +| HTTP handlers (`createHandler` default) | `10000` | +| Integration queue worker | `60000` | +| Outside an invocation (tests, scripts, `frigg start`) | `300000` | + +`remainingInvocationMs()` returns the milliseconds left, or `Infinity` outside +an invocation. `runWithInvocationDeadline({ deadlineAt, maxInProcessWaitMs }, fn)` +runs `fn` with either limit or both. `deadlineAt` is in epoch milliseconds. A +nested call can only lower each limit, field by field. A field that is not a +finite number, or a negative `maxInProcessWaitMs`, sets nothing. `0` means never +sleep: every hinted wait throws `RateLimitError`. + ## Complete OAuth2 Example ```javascript diff --git a/packages/core/CLAUDE.md b/packages/core/CLAUDE.md index 91b628746..5f827ca62 100644 --- a/packages/core/CLAUDE.md +++ b/packages/core/CLAUDE.md @@ -369,6 +369,16 @@ class MyIntegration extends IntegrationBase { - `ApiKeyRequester` - API key authentication - `BasicAuthRequester` - Basic authentication +**Rate limits** (ADR-049): an API module declares `static rateLimit` on its +`Requester` subclass. A hinted wait (from `classify()`, the header parsers or +the static policy) sleeps within a budget that the caller owns (10 s in an HTTP +handler, 60 s in a queue worker, 300 s outside an invocation, at most 10 +hinted retries), or throws `RateLimitError` +(`isRateLimited`, `retryAt`) when it does not fit. A 429 with no hint keeps the +fixed `backOff` ladder. A non-429 limit that `classify` names without a time +throws `RateLimitError` at once, with `retryAt` unknown. See +[Rate Limits](../../docs/reference/api-module-definition-and-functions.md#rate-limits-static-ratelimit). + **Module Factory**: - `ModuleFactory` - Creates and configures API module instances - Handles credential injection @@ -477,7 +487,7 @@ handlers (`{ req, res, next }`) and `this.on` events do not. `isLastAttempt` is `false` when either count is unknown: a local or non-SQS invocation, or a queue whose redrive policy the stack does not own (`ownership.queue: 'external'`). The value is information only: core still -rethrows retryable errors and discards halt errors (4xx except 408/429). +rethrows retryable errors and discards halt errors (4xx except 408 and 429). Use it to end a run or count lost work on the final try: when a retryable error (429, 5xx, network) is about to be rethrown and `isLastAttempt` is @@ -507,6 +517,7 @@ it as `FRIGG_QUEUE_MAX_RECEIVE_COUNT` on the queue worker function. **Error Types**: - `BaseError` - Base error class - `FetchError` - HTTP request failures +- `RateLimitError` - A `FetchError` for a limit the provider or the module's policy says when to retry (`retryAt`) - `HaltError` - Stop processing without retry - `RequiredPropertyError` - Missing required parameters - `ParameterTypeError` - Invalid parameter type diff --git a/packages/core/core/CLAUDE.md b/packages/core/core/CLAUDE.md index 520a70d48..905de2de5 100644 --- a/packages/core/core/CLAUDE.md +++ b/packages/core/core/CLAUDE.md @@ -130,7 +130,7 @@ class MyIntegration extends Delegate { ### Lambda Handler Lifecycle 1. **Pre-Execution Setup**: ```javascript - runInvocationScope({ requestId, handlerName, method, route, invocation }, …); // Logger scope for the invocation + runInvocationScope({ requestId, handlerName, method, route, invocation }, …); // Logger scope + invocation deadline (remainingInvocationMs()) + in-process wait budget (createHandler `maxInProcessWaitMs`: 10 s HTTP, 60 s queue worker) log.info('Handler invoked', { eventName: 'frigg.handler.invoked' }); await secretsToEnv(); // Secrets Manager injection await parametersToEnv(); // SSM Parameter Store fetch (only when SSM_PARAMETER_PREFIX + FRIGG_SSM_OFFLOADED_KEYS are set) @@ -226,6 +226,8 @@ const handler = createHandler({ - Logs error but returns success - Prevents infinite retries for known issues - Used for graceful degradation scenarios + - The queue worker sets it on a 4xx except 408, 429 and errors with + `isRateLimited` (a provider limit signalled with 403 clears with time) ### Logging Strategy ```javascript diff --git a/packages/core/core/create-handler.js b/packages/core/core/create-handler.js index e5061d765..855b6e5dd 100644 --- a/packages/core/core/create-handler.js +++ b/packages/core/core/create-handler.js @@ -12,6 +12,7 @@ const { runInvocationScope, DEFAULT_FLUSH_TIMEOUT_MS, } = require('./invocation-scope'); +const { MAX_IN_PROCESS_WAIT_MS } = require('./invocation-deadline'); const { secretsToEnv } = require('./secrets-to-env'); const { parametersToEnv } = require('./parameters-to-env'); const { @@ -31,6 +32,7 @@ const createHandler = (optionByName = {}) => { telemetry, flushTimeoutMs = DEFAULT_FLUSH_TIMEOUT_MS, usageRollup, + maxInProcessWaitMs = MAX_IN_PROCESS_WAIT_MS.http, } = optionByName; if (!method) { @@ -144,6 +146,7 @@ const createHandler = (optionByName = {}) => { shouldUseDatabase, flushTimeoutMs, context, + maxInProcessWaitMs, } ); }; diff --git a/packages/core/core/create-handler.test.js b/packages/core/core/create-handler.test.js index a3cf9678f..ba1e027b0 100644 --- a/packages/core/core/create-handler.test.js +++ b/packages/core/core/create-handler.test.js @@ -7,6 +7,7 @@ jest.mock('./secrets-to-env', () => ({ const { connectPrisma } = require('../database/prisma'); const { createHandler } = require('./create-handler'); +const { invocationMaxInProcessWaitMs } = require('./invocation-deadline'); const { getLogger, createMemorySink } = require('../logs'); const { HaltError } = require('../errors/halt-error'); const { httpApiV2Event, sqsEvent } = require('../logs/__fixtures__/events'); @@ -476,3 +477,26 @@ describe('createHandler — logger scope and records (ADR-048)', () => { expect(inside.droppedKeys).toEqual(['requestId']); }); }); + +describe('createHandler: in-process wait budget', () => { + const ctx = { awsRequestId: 'r1' }; + + it('runs the method with a 10 s budget by default', async () => { + const handler = createHandler({ + eventName: 'X', + shouldUseDatabase: false, + method: async () => invocationMaxInProcessWaitMs(), + }); + await expect(handler({}, { ...ctx })).resolves.toBe(10_000); + }); + + it('runs the method with the budget the caller gives', async () => { + const handler = createHandler({ + eventName: 'X', + shouldUseDatabase: false, + maxInProcessWaitMs: 60_000, + method: async () => invocationMaxInProcessWaitMs(), + }); + await expect(handler({}, { ...ctx })).resolves.toBe(60_000); + }); +}); diff --git a/packages/core/core/index.js b/packages/core/core/index.js index c69d9a2d4..f648c2e28 100644 --- a/packages/core/core/index.js +++ b/packages/core/core/index.js @@ -3,6 +3,10 @@ const { Worker } = require('./Worker'); const { loadInstalledModules } = require('./load-installed-modules'); const { createHandler } = require('./create-handler'); const { runInvocationScope } = require('./invocation-scope'); +const { + runWithInvocationDeadline, + remainingInvocationMs, +} = require('./invocation-deadline'); module.exports = { Delegate, @@ -10,4 +14,6 @@ module.exports = { loadInstalledModules, createHandler, runInvocationScope, + runWithInvocationDeadline, + remainingInvocationMs, }; diff --git a/packages/core/core/invocation-deadline.js b/packages/core/core/invocation-deadline.js new file mode 100644 index 000000000..5c8a6e13e --- /dev/null +++ b/packages/core/core/invocation-deadline.js @@ -0,0 +1,91 @@ +const { AsyncLocalStorage } = require('node:async_hooks'); + +const STORE_KEY = Symbol.for('@friggframework/core.invocation-deadline'); + +function storage() { + if (!globalThis[STORE_KEY]) { + globalThis[STORE_KEY] = new AsyncLocalStorage(); + } + return globalThis[STORE_KEY]; +} + +const MAX_IN_PROCESS_WAIT_MS = Object.freeze({ + outsideInvocation: 300_000, + http: 10_000, + queueWorker: 60_000, +}); + +const validDeadline = (value) => + typeof value === 'number' && Number.isFinite(value) ? value : undefined; + +const validWait = (value) => + typeof value === 'number' && Number.isFinite(value) && value >= 0 + ? value + : undefined; + +const lower = (outer, next) => + outer === undefined + ? next + : next === undefined + ? outer + : Math.min(outer, next); + +function runWithInvocationDeadline( + { deadlineAt, maxInProcessWaitMs } = {}, + fn +) { + const outer = storage().getStore(); + const effectiveDeadline = lower( + outer?.deadlineAt, + validDeadline(deadlineAt) + ); + const effectiveWait = lower( + outer?.maxInProcessWaitMs, + validWait(maxInProcessWaitMs) + ); + if (effectiveDeadline === undefined && effectiveWait === undefined) { + return fn(); + } + return storage().run( + Object.freeze({ + deadlineAt: effectiveDeadline ?? Infinity, + maxInProcessWaitMs: effectiveWait, + }), + fn + ); +} + +function remainingTimeMs(context) { + try { + const remaining = context?.getRemainingTimeInMillis?.(); + return typeof remaining === 'number' ? remaining : Infinity; + } catch (_) { + return Infinity; + } +} + +function deadlineFromContext(context) { + const remaining = remainingTimeMs(context); + return Number.isFinite(remaining) ? Date.now() + remaining : undefined; +} + +function remainingInvocationMs(now = Date.now()) { + const store = storage().getStore(); + return store ? Math.max(0, store.deadlineAt - now) : Infinity; +} + +function invocationMaxInProcessWaitMs() { + return ( + storage().getStore()?.maxInProcessWaitMs ?? + MAX_IN_PROCESS_WAIT_MS.outsideInvocation + ); +} + +module.exports = { + MAX_IN_PROCESS_WAIT_MS, + deadlineFromContext, + invocationMaxInProcessWaitMs, + remainingInvocationMs, + remainingTimeMs, + runWithInvocationDeadline, +}; diff --git a/packages/core/core/invocation-deadline.test.js b/packages/core/core/invocation-deadline.test.js new file mode 100644 index 000000000..49496d6ad --- /dev/null +++ b/packages/core/core/invocation-deadline.test.js @@ -0,0 +1,208 @@ +const { + MAX_IN_PROCESS_WAIT_MS, + deadlineFromContext, + invocationMaxInProcessWaitMs, + remainingInvocationMs, + runWithInvocationDeadline, +} = require('./invocation-deadline'); + +const tick = () => new Promise((resolve) => setImmediate(resolve)); + +describe('core/invocation-deadline', () => { + afterEach(() => jest.restoreAllMocks()); + + it('has no limit outside a deadline scope', () => { + expect(remainingInvocationMs()).toBe(Infinity); + }); + + it('reports the time left, deep inside async work', async () => { + const deadlineAt = 1_060_000; + await runWithInvocationDeadline({ deadlineAt }, async () => { + await tick(); + await Promise.resolve().then(() => { + expect(remainingInvocationMs(1_000_000)).toBe(60_000); + }); + }); + }); + + it('never reports a negative time', () => { + runWithInvocationDeadline({ deadlineAt: 1_000 }, () => { + expect(remainingInvocationMs(5_000)).toBe(0); + }); + }); + + it('returns the value of fn and passes a rejection through', async () => { + await expect( + runWithInvocationDeadline({ deadlineAt: 2e12 }, async () => 'ok') + ).resolves.toBe('ok'); + const boom = new Error('boom'); + await expect( + runWithInvocationDeadline({ deadlineAt: 2e12 }, async () => { + throw boom; + }) + ).rejects.toBe(boom); + }); + + it.each([undefined, null, NaN, Infinity, 'soon'])( + 'runs fn with no limit for the deadline %p', + (deadlineAt) => { + const result = runWithInvocationDeadline({ deadlineAt }, () => + remainingInvocationMs() + ); + expect(result).toBe(Infinity); + } + ); + + it('lets a nested scope shorten the deadline, then restores the outer one', () => { + runWithInvocationDeadline({ deadlineAt: 1_060_000 }, () => { + runWithInvocationDeadline({ deadlineAt: 1_010_000 }, () => { + expect(remainingInvocationMs(1_000_000)).toBe(10_000); + }); + expect(remainingInvocationMs(1_000_000)).toBe(60_000); + }); + }); + + it('does not let a nested scope extend the deadline', () => { + runWithInvocationDeadline({ deadlineAt: 1_010_000 }, () => { + runWithInvocationDeadline({ deadlineAt: 1_060_000 }, () => { + expect(remainingInvocationMs(1_000_000)).toBe(10_000); + }); + }); + }); + + describe('maxInProcessWaitMs', () => { + it('has one table of defaults, frozen', () => { + expect(MAX_IN_PROCESS_WAIT_MS).toEqual({ + outsideInvocation: 300_000, + http: 10_000, + queueWorker: 60_000, + }); + expect(Object.isFrozen(MAX_IN_PROCESS_WAIT_MS)).toBe(true); + }); + + it('reads the outside-invocation default when no scope is open', () => { + expect(invocationMaxInProcessWaitMs()).toBe(300_000); + }); + + it('reads the default when the scope sets only a deadline', () => { + runWithInvocationDeadline({ deadlineAt: 2e12 }, () => { + expect(invocationMaxInProcessWaitMs()).toBe(300_000); + }); + }); + + it('reads the value of the scope, deep inside async work', async () => { + await runWithInvocationDeadline( + { maxInProcessWaitMs: 10_000 }, + async () => { + await tick(); + expect(invocationMaxInProcessWaitMs()).toBe(10_000); + } + ); + }); + + it('sets no deadline when it is the only field', () => { + runWithInvocationDeadline({ maxInProcessWaitMs: 0 }, () => { + expect(invocationMaxInProcessWaitMs()).toBe(0); + expect(remainingInvocationMs()).toBe(Infinity); + }); + }); + + it.each([undefined, null, NaN, Infinity, -1, '5'])( + 'ignores the value %p and runs fn with no scope', + (maxInProcessWaitMs) => { + runWithInvocationDeadline({ maxInProcessWaitMs }, () => { + expect(invocationMaxInProcessWaitMs()).toBe(300_000); + expect(remainingInvocationMs()).toBe(Infinity); + }); + } + ); + + it('lets a nested scope lower the value, then restores the outer one', () => { + runWithInvocationDeadline({ maxInProcessWaitMs: 60_000 }, () => { + runWithInvocationDeadline({ maxInProcessWaitMs: 5_000 }, () => { + expect(invocationMaxInProcessWaitMs()).toBe(5_000); + }); + expect(invocationMaxInProcessWaitMs()).toBe(60_000); + }); + }); + + it('does not let a nested scope raise the value', () => { + runWithInvocationDeadline({ maxInProcessWaitMs: 5_000 }, () => { + runWithInvocationDeadline( + { maxInProcessWaitMs: 60_000 }, + () => { + expect(invocationMaxInProcessWaitMs()).toBe(5_000); + } + ); + }); + }); + + it('takes the lower of each field on its own', () => { + runWithInvocationDeadline( + { deadlineAt: 1_010_000, maxInProcessWaitMs: 60_000 }, + () => { + runWithInvocationDeadline( + { deadlineAt: 1_060_000, maxInProcessWaitMs: 5_000 }, + () => { + expect(remainingInvocationMs(1_000_000)).toBe( + 10_000 + ); + expect(invocationMaxInProcessWaitMs()).toBe(5_000); + } + ); + } + ); + }); + + it('keeps the outer value of a field the nested scope leaves out', () => { + runWithInvocationDeadline( + { deadlineAt: 1_010_000, maxInProcessWaitMs: 60_000 }, + () => { + runWithInvocationDeadline({ deadlineAt: 1_005_000 }, () => { + expect(invocationMaxInProcessWaitMs()).toBe(60_000); + }); + runWithInvocationDeadline( + { maxInProcessWaitMs: 5_000 }, + () => { + expect(remainingInvocationMs(1_000_000)).toBe( + 10_000 + ); + } + ); + } + ); + }); + }); + + describe('deadlineFromContext', () => { + it('adds the time left to the current time', () => { + jest.spyOn(Date, 'now').mockReturnValue(1_000); + expect( + deadlineFromContext({ getRemainingTimeInMillis: () => 5_000 }) + ).toBe(6_000); + }); + + it.each([ + ['no context', undefined], + ['a context without the method', {}], + [ + 'a time that is not a number', + { getRemainingTimeInMillis: () => 'soon' }, + ], + [ + 'a time that is not finite', + { getRemainingTimeInMillis: () => Infinity }, + ], + [ + 'a method that throws', + { + getRemainingTimeInMillis: () => { + throw new Error('gone'); + }, + }, + ], + ])('returns undefined for %s', (_label, context) => { + expect(deadlineFromContext(context)).toBeUndefined(); + }); + }); +}); diff --git a/packages/core/core/invocation-scope.js b/packages/core/core/invocation-scope.js index 93e85f95e..ccaa69646 100644 --- a/packages/core/core/invocation-scope.js +++ b/packages/core/core/invocation-scope.js @@ -1,6 +1,11 @@ const { runInContext, LOGGER_SCOPE_KEY } = require('../logs/context'); const { summarizeMessageBody } = require('../logs/summarize-event'); const { flushSinks, hasFlushableSinks } = require('../logs/logger-runtime'); +const { + deadlineFromContext, + remainingTimeMs, + runWithInvocationDeadline, +} = require('./invocation-deadline'); // Bounds the tail latency telemetry adds to every warm invocation. Kept low so // an unreachable OTLP endpoint (e.g. a VPC Lambda with no NAT/egress) costs at @@ -111,15 +116,6 @@ function withDeadline(ms, fn) { }); } -function remainingTimeMs(context) { - try { - const remaining = context?.getRemainingTimeInMillis?.(); - return typeof remaining === 'number' ? remaining : Infinity; - } catch (_) { - return Infinity; - } -} - async function flushInvocation({ telemetry, usageRollup, @@ -167,22 +163,32 @@ async function runInvocationScope(fields, fn, opts = {}) { shouldUseDatabase = true, flushTimeoutMs = DEFAULT_FLUSH_TIMEOUT_MS, context, + maxInProcessWaitMs, } = opts; - return runInContext({ [LOGGER_SCOPE_KEY]: { ...(fields || {}) } }, async () => { - try { - return await fn(); - } finally { - // Every step is guarded, so a flush never changes the result. - await flushInvocation({ - telemetry, - usageRollup, - eventSummary, - shouldUseDatabase, - flushTimeoutMs, - context, - }); - } - }); + const deadline = { + deadlineAt: deadlineFromContext(context), + maxInProcessWaitMs, + }; + return runWithInvocationDeadline(deadline, () => + runInContext( + { [LOGGER_SCOPE_KEY]: { ...(fields || {}) } }, + async () => { + try { + return await fn(); + } finally { + // Every step is guarded, so a flush never changes the result. + await flushInvocation({ + telemetry, + usageRollup, + eventSummary, + shouldUseDatabase, + flushTimeoutMs, + context, + }); + } + } + ) + ); } function toCount(value) { diff --git a/packages/core/core/invocation-scope.test.js b/packages/core/core/invocation-scope.test.js index 22edbcab1..6766778ce 100644 --- a/packages/core/core/invocation-scope.test.js +++ b/packages/core/core/invocation-scope.test.js @@ -10,6 +10,10 @@ const { const { getLogger, createMemorySink } = require('../logs'); const { setSinks } = require('../logs/logger-runtime'); const { getLoggerScope } = require('../logs/context'); +const { + invocationMaxInProcessWaitMs, + remainingInvocationMs, +} = require('./invocation-deadline'); const { mergeTelemetryContext } = require('../telemetry/telemetry-context'); const { NoOpTelemetry } = require('../telemetry/no-op-telemetry'); const { createUsageRollupSubscriber } = require('../telemetry/usage-rollup-subscriber'); @@ -59,6 +63,44 @@ describe('core/invocation-scope', () => { }); }); + it('makes the time left in the invocation visible deep inside fn', async () => { + const context = { getRemainingTimeInMillis: () => 60_000 }; + await runInvocationScope( + {}, + async () => { + await tick(); + await Promise.resolve().then(() => { + expect(remainingInvocationMs()).toBeLessThanOrEqual(60_000); + expect(remainingInvocationMs()).toBeGreaterThan(55_000); + }); + }, + { context } + ); + }); + + it('passes maxInProcessWaitMs on to the invocation scope', async () => { + await runInvocationScope( + {}, + async () => { + expect(invocationMaxInProcessWaitMs()).toBe(12_000); + expect(remainingInvocationMs()).toBe(Infinity); + }, + { maxInProcessWaitMs: 12_000 } + ); + }); + + it('sets the default wait budget when the caller gives none', async () => { + await runInvocationScope({}, async () => { + expect(invocationMaxInProcessWaitMs()).toBe(300_000); + }); + }); + + it('sets no deadline when there is no context', async () => { + await runInvocationScope({}, async () => { + expect(remainingInvocationMs()).toBe(Infinity); + }); + }); + it('drops undefined fields, so an absent context sets no requestId', async () => { await runInvocationScope({ requestId: undefined, handlerName: 'H' }, async () => { expect(getLoggerScope()).toEqual({ handlerName: 'H' }); diff --git a/packages/core/errors/fetch-error.js b/packages/core/errors/fetch-error.js index 545c8087a..b3b139109 100644 --- a/packages/core/errors/fetch-error.js +++ b/packages/core/errors/fetch-error.js @@ -55,12 +55,16 @@ class FetchError extends BaseError { static async create(options = {}) { const { response } = options; - let responseBody = - response && !response.bodyUsed && typeof response.text === 'function' - ? await response.text() - : null; - if (!responseBody) responseBody = options.responseBody ?? options.body; - return new FetchError({ ...options, responseBody }); + let { responseBody } = options; + if (responseBody === undefined || responseBody === null) { + const canRead = + response && + !response.bodyUsed && + typeof response.text === 'function'; + responseBody = (canRead && (await response.text())) || options.body; + } + const ErrorClass = typeof this === 'function' ? this : FetchError; + return new ErrorClass({ ...options, responseBody }); } } diff --git a/packages/core/errors/fetch-error.test.js b/packages/core/errors/fetch-error.test.js index bf7c8ba3c..b428cd372 100644 --- a/packages/core/errors/fetch-error.test.js +++ b/packages/core/errors/fetch-error.test.js @@ -192,6 +192,62 @@ describe('FetchError', () => { expect(error.body).toBe('fallback'); }); + it('create() reads the stream before options.body', async () => { + const text = jest.fn(async () => 'from the stream'); + const error = await FetchError.create({ + response: { status: 400, bodyUsed: false, text }, + body: 'fallback', + }); + expect(text).toHaveBeenCalledTimes(1); + expect(error.body).toBe('from the stream'); + }); + + it('create() falls back to options.body when the stream gives nothing', async () => { + const error = await FetchError.create({ + response: { status: 400, bodyUsed: false, text: async () => '' }, + body: 'fallback', + }); + expect(error.body).toBe('fallback'); + }); + + it('create() uses a given responseBody and does not read the stream again', async () => { + const text = jest.fn(async () => 'second read'); + const error = await FetchError.create({ + response: { status: 429, bodyUsed: false, text }, + responseBody: 'already read', + }); + expect(text).not.toHaveBeenCalled(); + expect(error.body).toBe('already read'); + }); + + it('create() keeps an empty responseBody it is given and does not read the stream', async () => { + const text = jest.fn(async () => 'second read'); + const error = await FetchError.create({ + response: { status: 500, bodyUsed: false, text }, + responseBody: '', + }); + expect(text).not.toHaveBeenCalled(); + expect(error.body).toBe(''); + }); + + it('create() constructs the subclass it is called on', async () => { + class ProviderError extends FetchError {} + const error = await ProviderError.create({ + resource: 'https://h.example', + response: { status: 400, bodyUsed: true }, + }); + expect(error).toBeInstanceOf(ProviderError); + expect(error.name).toBe('ProviderError'); + }); + + it('create() still builds a FetchError when called without its class', async () => { + const { create } = FetchError; + const error = await create({ + response: { status: 400, bodyUsed: true }, + }); + expect(error).toBeInstanceOf(FetchError); + }); + it('create() propagates a rejected body read', async () => { const aborted = new Error('aborted mid-body'); aborted.name = 'AbortError'; diff --git a/packages/core/errors/index.js b/packages/core/errors/index.js index 836274679..b224b2cf1 100644 --- a/packages/core/errors/index.js +++ b/packages/core/errors/index.js @@ -1,6 +1,7 @@ const { BaseError } = require('./base-error'); const { FetchError } = require('./fetch-error'); const { HaltError } = require('./halt-error'); +const { RateLimitError } = require('./rate-limit-error'); const { RequiredPropertyError, ParameterTypeError, @@ -11,6 +12,7 @@ module.exports = { BaseError, FetchError, HaltError, + RateLimitError, RequiredPropertyError, ParameterTypeError, ClientSafeError, diff --git a/packages/core/errors/rate-limit-error.js b/packages/core/errors/rate-limit-error.js new file mode 100644 index 000000000..a4b3808c7 --- /dev/null +++ b/packages/core/errors/rate-limit-error.js @@ -0,0 +1,46 @@ +const { FetchError } = require('./fetch-error'); + +function resolveTiming({ hint, waitMs, now }) { + if (waitMs !== undefined) { + return { waitMs, retryAt: new Date(now + waitMs) }; + } + if (hint?.retryAt) { + const retryAt = new Date(hint.retryAt); + if (!Number.isNaN(retryAt.getTime())) { + return { waitMs: Math.max(0, retryAt.getTime() - now), retryAt }; + } + } + if (hint?.waitMs !== undefined) { + return { + waitMs: hint.waitMs, + retryAt: new Date(now + hint.waitMs), + }; + } + return { waitMs: undefined, retryAt: undefined }; +} + +class RateLimitError extends FetchError { + constructor({ + hint, + waitMs, + module, + scopeKey, + now = Date.now(), + ...fetchErrorArgs + } = {}) { + super(fetchErrorArgs); + + const timing = resolveTiming({ hint, waitMs, now }); + this.statusCode ??= fetchErrorArgs.cause?.statusCode; + this.isRateLimited = true; + this.waitMs = timing.waitMs; + this.retryAt = timing.retryAt; + this.reason = hint?.reason ?? 'unknown'; + this.policy = hint?.policy; + this.source = hint?.source ?? 'unknown'; + this.module = module; + this.scopeKey = scopeKey; + } +} + +module.exports = { RateLimitError }; diff --git a/packages/core/errors/rate-limit-error.test.js b/packages/core/errors/rate-limit-error.test.js new file mode 100644 index 000000000..285304c3c --- /dev/null +++ b/packages/core/errors/rate-limit-error.test.js @@ -0,0 +1,248 @@ +const util = require('node:util'); +const { BaseError } = require('./base-error'); +const { FetchError } = require('./fetch-error'); +const errors = require('./index'); +const { RateLimitError } = require('./rate-limit-error'); +const { serializeError } = require('../logs'); +const { SECRETS } = require('../logs/__fixtures__/secrets'); +const { toContainNoSecretWindow } = require('../logs/__fixtures__/matchers'); + +expect.extend({ toContainNoSecretWindow }); + +const NOW = Date.parse('2026-09-28T12:00:00.000Z'); + +const secretUrl = `https://user:${SECRETS.password}@api.example.com/v1/items?api_key=${SECRETS.apiKeyQuery}&page=2`; +const sanitizedUrl = + 'https://api.example.com/v1/items?api_key=REDACTED&page=REDACTED'; + +function secretInit() { + return { + method: 'POST', + headers: { Authorization: `Bearer ${SECRETS.bearer}` }, + body: `client_secret=${SECRETS.clientSecret}`, + }; +} + +function throttledResponse(overrides = {}) { + return { + status: 429, + bodyUsed: false, + text: jest.fn(async () => `{"access_token":"${SECRETS.accessToken}"}`), + ...overrides, + }; +} + +const hint = { + waitMs: 5_000, + retryAt: new Date(NOW + 5_000), + reason: 'daily', + policy: 'DAILY', + source: 'header', +}; + +describe('RateLimitError', () => { + it('is a FetchError and a BaseError named RateLimitError', () => { + const error = new RateLimitError({ hint, now: NOW }); + expect(error).toBeInstanceOf(FetchError); + expect(error).toBeInstanceOf(BaseError); + expect(error.name).toBe('RateLimitError'); + }); + + it('is exported from the errors index', () => { + expect(errors.RateLimitError).toBe(RateLimitError); + }); + + it('can be built with no arguments', () => { + const error = new RateLimitError(); + expect(error.isRateLimited).toBe(true); + expect(error.waitMs).toBeUndefined(); + expect(error.retryAt).toBeUndefined(); + expect(error.reason).toBe('unknown'); + expect(error.source).toBe('unknown'); + }); + + it('flags itself as rate limited', () => { + expect(new RateLimitError({ hint }).isRateLimited).toBe(true); + }); + + describe('retryAt and waitMs', () => { + it('counts a given waitMs from now', () => { + const error = new RateLimitError({ hint, waitMs: 8_000, now: NOW }); + expect(error.waitMs).toBe(8_000); + expect(error.retryAt).toEqual(new Date(NOW + 8_000)); + }); + + it('keeps the retryAt of the hint when no waitMs is given', () => { + const error = new RateLimitError({ hint, now: NOW + 1_000 }); + expect(error.retryAt).toEqual(hint.retryAt); + expect(error.waitMs).toBe(4_000); + }); + + it('waits 0 ms when the retryAt of the hint has passed', () => { + const error = new RateLimitError({ hint, now: NOW + 60_000 }); + expect(error.retryAt).toEqual(hint.retryAt); + expect(error.waitMs).toBe(0); + }); + + it('uses the waitMs of a hint that has no retryAt', () => { + const error = new RateLimitError({ + hint: { waitMs: 2_000 }, + now: NOW, + }); + expect(error.waitMs).toBe(2_000); + expect(error.retryAt).toEqual(new Date(NOW + 2_000)); + }); + + it('ignores a retryAt that is not a valid date', () => { + const error = new RateLimitError({ + hint: { retryAt: 'never', waitMs: 3_000 }, + now: NOW, + }); + expect(error.waitMs).toBe(3_000); + expect(error.retryAt).toEqual(new Date(NOW + 3_000)); + }); + + it('leaves retryAt and waitMs unknown for a hint that carries no time', () => { + const error = new RateLimitError({ + hint: { reason: 'daily' }, + now: NOW, + }); + expect(error.retryAt).toBeUndefined(); + expect(error.waitMs).toBeUndefined(); + expect(error.reason).toBe('daily'); + }); + + it('reads the clock when no time is given', () => { + const before = Date.now(); + const error = new RateLimitError({ waitMs: 1_000 }); + expect(error.retryAt.getTime()).toBeGreaterThanOrEqual( + before + 1_000 + ); + }); + }); + + describe('fields', () => { + it('carries the reason, policy, source, module and scope key', async () => { + const error = await RateLimitError.create({ + resource: secretUrl, + init: secretInit(), + response: throttledResponse(), + hint, + waitMs: 5_000, + module: 'hubspot', + scopeKey: 'hubspot:entity:7', + }); + expect(error).toMatchObject({ + isRateLimited: true, + statusCode: 429, + method: 'POST', + url: sanitizedUrl, + waitMs: 5_000, + reason: 'daily', + policy: 'DAILY', + source: 'header', + module: 'hubspot', + scopeKey: 'hubspot:entity:7', + }); + expect(error.retryAt).toBeInstanceOf(Date); + expect(error.message).toBe(`POST ${sanitizedUrl} 429`); + }); + + it('takes the status of the cause when there is no response', () => { + const cause = Object.assign(new Error('REQUEST_LIMIT_EXCEEDED'), { + statusCode: 403, + }); + const error = new RateLimitError({ hint, cause }); + expect(error.statusCode).toBe(403); + }); + + it('keeps the status of the response over the status of the cause', () => { + const cause = Object.assign(new Error('boom'), { statusCode: 403 }); + const error = new RateLimitError({ + hint, + cause, + response: throttledResponse(), + }); + expect(error.statusCode).toBe(429); + }); + + it('does not put the hint in the enumerable fields', () => { + const error = new RateLimitError({ hint }); + expect(Object.keys(error)).not.toContain('hint'); + expect(Object.keys(error)).not.toContain('hints'); + }); + }); + + describe('create()', () => { + it('reads the body once and keeps it out of the enumerable fields', async () => { + const response = throttledResponse(); + const error = await RateLimitError.create({ + resource: 'https://h.example', + response, + hint, + }); + expect(response.text).toHaveBeenCalledTimes(1); + expect(error).toBeInstanceOf(RateLimitError); + expect(error.body).toContain('access_token'); + expect(Object.keys(error)).not.toContain('body'); + }); + + it('does not read the stream when the body was read already', async () => { + const response = throttledResponse(); + const error = await RateLimitError.create({ + resource: 'https://h.example', + response, + responseBody: '{"policyName":"DAILY"}', + hint, + }); + expect(response.text).not.toHaveBeenCalled(); + expect(error.body).toBe('{"policyName":"DAILY"}'); + }); + }); + + describe('secrets', () => { + it('leaves no secret in the message, the stack, util.inspect or JSON', async () => { + const error = await RateLimitError.create({ + resource: secretUrl, + init: secretInit(), + response: throttledResponse(), + hint, + module: 'hubspot', + }); + expect(error.message).toContainNoSecretWindow(SECRETS); + expect(error.stack).toContainNoSecretWindow(SECRETS); + expect(util.inspect(error, { depth: 5 })).toContainNoSecretWindow( + SECRETS + ); + expect(JSON.stringify(error)).toContainNoSecretWindow(SECRETS); + }); + }); + + describe('logging', () => { + it('serializes with its own type and the status', async () => { + const error = await RateLimitError.create({ + resource: 'https://h.example', + response: throttledResponse(), + hint, + }); + expect(serializeError(error)).toMatchObject({ + type: 'RateLimitError', + status: 429, + }); + }); + + it('serializes when the limit clears and why', () => { + const error = new RateLimitError({ + resource: 'https://h.example', + response: { status: 429 }, + hint, + waitMs: 5_000, + now: NOW, + }); + expect(serializeError(error)).toMatchObject({ + retryAt: '2026-09-28T12:00:05.000Z', + reason: 'daily', + }); + }); + }); +}); diff --git a/packages/core/handlers/workers/integration-defined-workers-budget.test.js b/packages/core/handlers/workers/integration-defined-workers-budget.test.js new file mode 100644 index 000000000..4650016ae --- /dev/null +++ b/packages/core/handlers/workers/integration-defined-workers-budget.test.js @@ -0,0 +1,26 @@ +const mockCreateHandler = jest.fn((options) => options); + +jest.mock('@friggframework/core', () => ({ + createHandler: mockCreateHandler, +})); +jest.mock('../app-definition-loader', () => ({ + loadAppDefinition: () => ({ + integrations: [{ Definition: { name: 'hubspot' } }], + }), +})); +jest.mock('../backend-utils', () => ({ + createQueueWorker: jest.fn(() => class {}), +})); + +const { MAX_IN_PROCESS_WAIT_MS } = require('../../core/invocation-deadline'); + +describe('integration queue worker handler', () => { + it('runs with the queue worker wait budget', () => { + const { handlers } = require('./integration-defined-workers'); + + expect(handlers.hubspot.queueWorker.maxInProcessWaitMs).toBe( + MAX_IN_PROCESS_WAIT_MS.queueWorker + ); + expect(MAX_IN_PROCESS_WAIT_MS.queueWorker).toBe(60_000); + }); +}); diff --git a/packages/core/handlers/workers/integration-defined-workers.js b/packages/core/handlers/workers/integration-defined-workers.js index cb88ac0b6..4d4ddcb12 100644 --- a/packages/core/handlers/workers/integration-defined-workers.js +++ b/packages/core/handlers/workers/integration-defined-workers.js @@ -1,6 +1,7 @@ const { createHandler } = require('@friggframework/core'); const { loadAppDefinition } = require('../app-definition-loader'); const { createQueueWorker } = require('../backend-utils'); +const { MAX_IN_PROCESS_WAIT_MS } = require('../../core/invocation-deadline'); // TODO(Phase 2): mount extension-declared workers in addition to the per-integration // default queue worker. Today, getExtensionWorkers(IntegrationClass) returns the // declared workers but they are not yet bound to dedicated SQS sources. Extension- @@ -19,6 +20,7 @@ integrationClasses.forEach((IntegrationClass) => { queueWorker: createHandler({ eventName: `Queue Worker for ${IntegrationClass.Definition.name}`, isUserFacingResponse: false, + maxInProcessWaitMs: MAX_IN_PROCESS_WAIT_MS.queueWorker, method: async (event, context) => { const worker = new defaultQueueWorker(); return await worker.run(event, context); diff --git a/packages/core/index.js b/packages/core/index.js index 30a91bbdd..7449ae984 100644 --- a/packages/core/index.js +++ b/packages/core/index.js @@ -12,6 +12,8 @@ const { loadInstalledModules, createHandler, runInvocationScope, + runWithInvocationDeadline, + remainingInvocationMs, } = require('./core/index'); const { prisma, @@ -55,6 +57,7 @@ const { BaseError, FetchError, HaltError, + RateLimitError, RequiredPropertyError, ParameterTypeError, } = require('./errors/index'); @@ -100,6 +103,12 @@ const { ModuleConstants, ModuleFactory, } = require('./modules/index'); +const { + classifyRateLimit, + parseRetryAfter, + parseResetHeaders, + parseIetfRateLimit, +} = require('./modules/requester/rate-limit'); const application = require('./application'); const utils = require('./utils'); @@ -120,6 +129,8 @@ module.exports = { loadInstalledModules, createHandler, runInvocationScope, + runWithInvocationDeadline, + remainingInvocationMs, // database prisma, @@ -142,9 +153,16 @@ module.exports = { BaseError, FetchError, HaltError, + RateLimitError, RequiredPropertyError, ParameterTypeError, + // rate limit + classifyRateLimit, + parseRetryAfter, + parseResetHeaders, + parseIetfRateLimit, + // integrations IntegrationBase, Options, diff --git a/packages/core/logs/serialize.js b/packages/core/logs/serialize.js index b85efc3fc..5034cf5c6 100644 --- a/packages/core/logs/serialize.js +++ b/packages/core/logs/serialize.js @@ -235,6 +235,20 @@ function errorCode(err) { return undefined; } +function rateLimitFields(err) { + const retryAt = attempt( + () => (err.isRateLimited ? err.retryAt : undefined), + undefined + ); + if (!(retryAt instanceof Date) || Number.isNaN(retryAt.getTime())) { + return {}; + } + const reason = attempt(() => err.reason, undefined); + return typeof reason === 'string' + ? { retryAt: retryAt.toISOString(), reason: cleanString(reason) } + : { retryAt: retryAt.toISOString() }; +} + function errorStack(err, type, message) { const stack = attempt(() => err.stack, undefined); if (typeof stack !== 'string') return undefined; @@ -279,6 +293,7 @@ function serializeErrorAt(err, depth, seen) { if (code !== undefined) out.code = code; const status = errorStatus(err); if (status !== undefined) out.status = status; + Object.assign(out, rateLimitFields(err)); const stack = errorStack(err, type, message); if (stack !== undefined) out.stack = stack; diff --git a/packages/core/logs/serialize.test.js b/packages/core/logs/serialize.test.js index d2aad21e3..5fae77f2c 100644 --- a/packages/core/logs/serialize.test.js +++ b/packages/core/logs/serialize.test.js @@ -51,6 +51,44 @@ describe('serializeError', () => { expect(serializeError(bare).type).toBe('Error'); }); + it('adds retryAt as ISO and reason for a rate-limited error with a valid retryAt', () => { + const err = Object.assign(new Error('GET https://h.example 429'), { + statusCode: 429, + isRateLimited: true, + retryAt: new Date('2026-09-28T12:00:05.000Z'), + reason: 'daily', + }); + expect(serializeError(err)).toMatchObject({ + status: 429, + retryAt: '2026-09-28T12:00:05.000Z', + reason: 'daily', + }); + }); + + it.each([ + [ + 'is not rate limited', + { retryAt: new Date('2026-09-28T12:00:05.000Z'), reason: 'daily' }, + ], + ['has no retryAt', { isRateLimited: true, reason: 'daily' }], + [ + 'has an invalid retryAt', + { isRateLimited: true, retryAt: new Date(NaN), reason: 'daily' }, + ], + [ + 'has a retryAt that is not a Date', + { + isRateLimited: true, + retryAt: '2026-09-28T12:00:05.000Z', + reason: 'daily', + }, + ], + ])('adds neither retryAt nor reason when the error %s', (_label, fields) => { + const out = serializeError(Object.assign(new Error('x'), fields)); + expect(out).not.toHaveProperty('retryAt'); + expect(out).not.toHaveProperty('reason'); + }); + it('reads status from statusCode, then status, then response.status', () => { const a = Object.assign(new Error('a'), { statusCode: 401, status: 500 }); const b = Object.assign(new Error('b'), { status: 404 }); diff --git a/packages/core/modules/requester/rate-limit/index.js b/packages/core/modules/requester/rate-limit/index.js new file mode 100644 index 000000000..316e2d957 --- /dev/null +++ b/packages/core/modules/requester/rate-limit/index.js @@ -0,0 +1,4 @@ +const parsers = require('./parsers'); +const policy = require('./policy'); + +module.exports = { ...parsers, ...policy }; diff --git a/packages/core/modules/requester/rate-limit/parsers.js b/packages/core/modules/requester/rate-limit/parsers.js new file mode 100644 index 000000000..dcf45190f --- /dev/null +++ b/packages/core/modules/requester/rate-limit/parsers.js @@ -0,0 +1,245 @@ +const MAX_HINT_WAIT_MS = 366 * 24 * 60 * 60 * 1000; + +const SECONDS_PATTERN = /^\d+(?:\.\d+)?$/; +const HAS_LETTER = /[A-Za-z]/; +const EPOCH_SECONDS_MIN = 1e9; +const EPOCH_MILLISECONDS_MIN = 1e12; + +const RESET_HEADERS = [ + 'x-ratelimit-reset', + 'ratelimit-reset', + 'x-rate-limit-reset', +]; +const REMAINING_HEADERS = [ + 'x-ratelimit-remaining', + 'ratelimit-remaining', + 'x-rate-limit-remaining', +]; + +const isMissing = (value) => value === undefined || value === null; + +function readEntries(headers, wanted) { + for (const entry of headers) { + if (Array.isArray(entry) && String(entry[0]).toLowerCase() === wanted) { + return entry[1]; + } + } + return undefined; +} + +function readKeys(headers, wanted) { + for (const key of Object.keys(headers)) { + if (key.toLowerCase() === wanted) return headers[key]; + } + return undefined; +} + +function headerValue(headers, name) { + if (!headers || typeof headers !== 'object') return undefined; + const wanted = name.toLowerCase(); + const hasGet = typeof headers.get === 'function'; + + let value; + if (hasGet) { + value = headers.get(name); + if (isMissing(value)) value = headers.get(wanted); + } + if (isMissing(value) && typeof headers[Symbol.iterator] === 'function') { + value = readEntries(headers, wanted); + } + if (isMissing(value) && !hasGet && !Array.isArray(headers)) { + value = readKeys(headers, wanted); + } + if (Array.isArray(value)) value = value[0]; + if (isMissing(value)) return undefined; + + const text = String(value).trim(); + return text === '' ? undefined : text; +} + +function definedOnly(fields) { + const result = {}; + for (const [key, value] of Object.entries(fields)) { + if (value !== undefined) result[key] = value; + } + return result; +} + +function hintFromWait(waitMs, now, { source = 'header', ...extra } = {}) { + if (!Number.isFinite(waitMs) || waitMs < 0 || waitMs > MAX_HINT_WAIT_MS) { + return null; + } + return { + waitMs, + retryAt: new Date(now + waitMs), + ...definedOnly(extra), + source, + }; +} + +function hintFromRetryAt(retryAt, now, { source = 'header', ...extra } = {}) { + const at = retryAt instanceof Date ? retryAt.getTime() : Number.NaN; + if (Number.isNaN(at)) return null; + const waitMs = Math.max(0, at - now); + if (waitMs > MAX_HINT_WAIT_MS) return null; + return { waitMs, retryAt: new Date(at), ...definedOnly(extra), source }; +} + +function parseDate(text) { + if (!HAS_LETTER.test(text)) return null; + const at = Date.parse(text); + return Number.isNaN(at) ? null : new Date(at); +} + +function parseRetryAfter(value, { now = Date.now() } = {}) { + if (isMissing(value)) return null; + const text = String(value).trim(); + if (text === '') return null; + if (SECONDS_PATTERN.test(text)) + return hintFromWait(Number(text) * 1000, now); + const date = parseDate(text); + return date ? hintFromRetryAt(date, now) : null; +} + +function numberOrUndefined(text) { + if (text === undefined) return undefined; + const number = Number(text); + return Number.isFinite(number) ? number : undefined; +} + +function firstHeader(headers, names) { + for (const name of names) { + const value = headerValue(headers, name); + if (value !== undefined) return value; + } + return undefined; +} + +function parseResetValue(text, now, extra) { + if (SECONDS_PATTERN.test(text)) { + const number = Number(text); + if (number >= EPOCH_MILLISECONDS_MIN) { + return hintFromRetryAt(new Date(number), now, extra); + } + if (number >= EPOCH_SECONDS_MIN) { + return hintFromRetryAt(new Date(number * 1000), now, extra); + } + return hintFromWait(number * 1000, now, extra); + } + const date = parseDate(text); + return date ? hintFromRetryAt(date, now, extra) : null; +} + +function parseResetHeaders(headers, { now = Date.now() } = {}) { + const remaining = numberOrUndefined( + firstHeader(headers, REMAINING_HEADERS) + ); + const extra = { remaining }; + + const after = headerValue(headers, 'x-ratelimit-reset-after'); + if (after !== undefined && SECONDS_PATTERN.test(after)) { + return hintFromWait(Number(after) * 1000, now, extra); + } + + for (const name of RESET_HEADERS) { + const value = headerValue(headers, name); + if (value === undefined) continue; + const hint = parseResetValue(value, now, extra); + if (hint) return hint; + } + return null; +} + +function splitList(text) { + const items = []; + let current = ''; + let quoted = false; + for (const char of text) { + if (char === '"') quoted = !quoted; + if (char === ',' && !quoted) { + items.push(current); + current = ''; + } else { + current += char; + } + } + items.push(current); + return items.map((item) => item.trim()).filter(Boolean); +} + +function parseStructuredItem(item) { + const [head, ...params] = item.split(';').map((part) => part.trim()); + const values = {}; + for (const param of params) { + const [key, raw] = param.split('=').map((part) => part.trim()); + if (key) values[key] = numberOrUndefined(raw); + } + const name = head.replace(/^"|"$/g, ''); + return { + name: name || undefined, + remaining: values.r, + waitSeconds: values.t, + }; +} + +function pickStructured(items) { + const usable = items.filter( + (item) => Number.isFinite(item.waitSeconds) && item.waitSeconds >= 0 + ); + usable.sort((a, b) => { + const byRemaining = + (a.remaining ?? Infinity) - (b.remaining ?? Infinity); + if (byRemaining !== 0 && !Number.isNaN(byRemaining)) return byRemaining; + return b.waitSeconds - a.waitSeconds; + }); + return usable[0]; +} + +function readKeyValue(text, key) { + const match = new RegExp( + String.raw`(?:^|[,;\s])${key}\s*=\s*([^,;\s]+)`, + 'i' + ).exec(text); + return match ? match[1] : undefined; +} + +function parseIetfRateLimit(headers, { now = Date.now() } = {}) { + const raw = headerValue(headers, 'ratelimit'); + if (raw === undefined) return null; + const policyHeader = headerValue(headers, 'ratelimit-policy'); + + if (/;\s*t\s*=/.test(raw)) { + const chosen = pickStructured(splitList(raw).map(parseStructuredItem)); + if (!chosen) return null; + return hintFromWait(chosen.waitSeconds * 1000, now, { + remaining: chosen.remaining, + policy: chosen.name ?? policyHeader, + }); + } + + const reset = readKeyValue(raw, 'reset'); + if (reset === undefined || !SECONDS_PATTERN.test(reset)) return null; + return hintFromWait(Number(reset) * 1000, now, { + remaining: numberOrUndefined(readKeyValue(raw, 'remaining')), + policy: policyHeader, + }); +} + +const BUILT_IN_PARSERS = { + retryAfter: (headers, { now } = {}) => + parseRetryAfter(headerValue(headers, 'retry-after'), { now }), + resetHeaders: parseResetHeaders, + ietf: parseIetfRateLimit, +}; + +module.exports = { + BUILT_IN_PARSERS, + MAX_HINT_WAIT_MS, + definedOnly, + headerValue, + hintFromRetryAt, + hintFromWait, + parseIetfRateLimit, + parseResetHeaders, + parseRetryAfter, +}; diff --git a/packages/core/modules/requester/rate-limit/parsers.test.js b/packages/core/modules/requester/rate-limit/parsers.test.js new file mode 100644 index 000000000..5192688ff --- /dev/null +++ b/packages/core/modules/requester/rate-limit/parsers.test.js @@ -0,0 +1,356 @@ +const { + BUILT_IN_PARSERS, + MAX_HINT_WAIT_MS, + headerValue, + parseIetfRateLimit, + parseResetHeaders, + parseRetryAfter, +} = require('./parsers'); + +const NOW = Date.parse('2026-09-28T12:00:00.000Z'); +const MAX_SECONDS = MAX_HINT_WAIT_MS / 1000; + +describe('rate-limit/parsers', () => { + describe('headerValue', () => { + const cases = { + 'a Headers-like object': () => ({ + get: (name) => + name.toLowerCase() === 'retry-after' ? '5' : null, + }), + 'a Map with a differently cased key': () => + new Map([['Retry-After', '5']]), + 'an entries array': () => [['Retry-After', '5']], + 'a plain object with a differently cased key': () => ({ + 'RETRY-AFTER': '5', + }), + 'a Headers-like object that is also iterable': () => ({ + get: () => null, + [Symbol.iterator]: function* () { + yield ['retry-after', '5']; + }, + }), + }; + + it.each(Object.keys(cases))('reads a header from %s', (shape) => { + expect(headerValue(cases[shape](), 'retry-after')).toBe('5'); + }); + + it.each([ + ['null headers', null], + ['undefined headers', undefined], + ['a string', 'Retry-After: 5'], + ['an empty object', {}], + ['a get() that finds nothing', { get: () => null }], + ])('returns undefined for %s', (_label, headers) => { + expect(headerValue(headers, 'retry-after')).toBeUndefined(); + }); + + it('trims the value and treats a blank value as missing', () => { + expect(headerValue({ 'retry-after': ' 7 ' }, 'retry-after')).toBe( + '7' + ); + expect( + headerValue({ 'retry-after': ' ' }, 'retry-after') + ).toBeUndefined(); + }); + + it('takes the first value of a multi-value header', () => { + expect( + headerValue({ 'retry-after': ['9', '10'] }, 'retry-after') + ).toBe('9'); + }); + }); + + describe('parseRetryAfter', () => { + it.each([ + ['120', 120_000], + ['0', 0], + ['1.5', 1_500], + [' 7 ', 7_000], + [String(MAX_SECONDS), MAX_HINT_WAIT_MS], + ])('reads delta seconds %j as %d ms', (value, waitMs) => { + const hint = parseRetryAfter(value, { now: NOW }); + expect(hint).toEqual({ + waitMs, + retryAt: new Date(NOW + waitMs), + source: 'header', + }); + }); + + it('reads an HTTP-date in the future', () => { + const value = new Date(NOW + 30_000).toUTCString(); + expect(parseRetryAfter(value, { now: NOW })).toEqual({ + waitMs: 30_000, + retryAt: new Date(NOW + 30_000), + source: 'header', + }); + }); + + it('reads an ISO timestamp in the future', () => { + const value = new Date(NOW + 45_000).toISOString(); + expect(parseRetryAfter(value, { now: NOW })).toEqual({ + waitMs: 45_000, + retryAt: new Date(NOW + 45_000), + source: 'header', + }); + }); + + it('keeps the date but waits 0 ms for a date in the past', () => { + const past = new Date(NOW - 60_000); + expect(parseRetryAfter(past.toUTCString(), { now: NOW })).toEqual({ + waitMs: 0, + retryAt: past, + source: 'header', + }); + }); + + it.each([ + ['a negative number', '-5'], + ['a word', 'soon'], + ['an empty string', ''], + ['undefined', undefined], + ['null', null], + ['a number that is not finite', '1e400'], + ['seconds above the cap', String(MAX_SECONDS + 1)], + ['a hex number', '0x10'], + ['a digit string with a suffix', '12abc'], + ['a bare ISO date without a time', '2026-09-28'], + ])('returns null for %s', (_label, value) => { + expect(parseRetryAfter(value, { now: NOW })).toBeNull(); + }); + + it('returns null for an HTTP-date above the cap', () => { + const far = new Date(NOW + MAX_HINT_WAIT_MS + 60_000).toUTCString(); + expect(parseRetryAfter(far, { now: NOW })).toBeNull(); + }); + + it('uses the current time when no clock is given', () => { + const before = Date.now(); + const hint = parseRetryAfter('10'); + expect(hint.retryAt.getTime()).toBeGreaterThanOrEqual( + before + 10_000 + ); + }); + }); + + describe('parseResetHeaders', () => { + const epochSeconds = (ms) => String(Math.floor((NOW + ms) / 1000)); + + it.each([ + [ + 'x-ratelimit-reset', + epochSeconds(90_000), + 90_000, + 'epoch seconds', + ], + [ + 'x-ratelimit-reset', + String(NOW + 90_000), + 90_000, + 'epoch milliseconds', + ], + ['x-ratelimit-reset', '60', 60_000, 'delta seconds'], + [ + 'x-ratelimit-reset', + new Date(NOW + 15_000).toISOString(), + 15_000, + 'an ISO timestamp', + ], + [ + 'x-rate-limit-reset', + epochSeconds(20_000), + 20_000, + 'epoch seconds (x-rate-limit-reset)', + ], + [ + 'ratelimit-reset', + '12', + 12_000, + 'delta seconds (ratelimit-reset)', + ], + [ + 'x-ratelimit-reset-after', + '2.5', + 2_500, + 'fractional delta seconds', + ], + ])('reads %s = %s as %d ms (%s)', (name, value, waitMs) => { + expect(parseResetHeaders({ [name]: value }, { now: NOW })).toEqual({ + waitMs, + retryAt: new Date(NOW + waitMs), + source: 'header', + }); + }); + + it('waits 0 ms when the reset time is already past', () => { + const hint = parseResetHeaders( + { 'x-ratelimit-reset': epochSeconds(-30_000) }, + { now: NOW } + ); + expect(hint.waitMs).toBe(0); + expect(hint.retryAt).toEqual(new Date(NOW - 30_000)); + }); + + it('prefers reset-after over a reset time', () => { + const hint = parseResetHeaders( + { + 'x-ratelimit-reset': epochSeconds(90_000), + 'x-ratelimit-reset-after': '3', + }, + { now: NOW } + ); + expect(hint.waitMs).toBe(3_000); + }); + + it('carries the remaining count', () => { + const hint = parseResetHeaders( + { 'x-ratelimit-reset': '10', 'x-ratelimit-remaining': '0' }, + { now: NOW } + ); + expect(hint.remaining).toBe(0); + }); + + it('ignores a remaining count that is not a number', () => { + const hint = parseResetHeaders( + { 'x-ratelimit-reset': '10', 'x-ratelimit-remaining': 'many' }, + { now: NOW } + ); + expect(hint.remaining).toBeUndefined(); + }); + + it.each([ + ['no reset header', { 'x-ratelimit-remaining': '0' }], + [ + 'a reset value that is not a time', + { 'x-ratelimit-reset': 'soon' }, + ], + ['a negative reset value', { 'x-ratelimit-reset': '-4' }], + ['a reset above the cap', { 'x-ratelimit-reset': '99999999' }], + ['no headers at all', undefined], + ])('returns null for %s', (_label, headers) => { + expect(parseResetHeaders(headers, { now: NOW })).toBeNull(); + }); + + it.each([ + [ + 'a Headers-like object', + { + get: (n) => + ({ 'x-ratelimit-reset': '30' }[n.toLowerCase()] ?? + null), + }, + ], + ['a Map', new Map([['X-RateLimit-Reset', '30']])], + ['an entries array', [['X-RateLimit-Reset', '30']]], + ['a plain object', { 'X-RateLimit-Reset': '30' }], + ])('finds the header on %s', (_label, headers) => { + expect(parseResetHeaders(headers, { now: NOW }).waitMs).toBe( + 30_000 + ); + }); + }); + + describe('parseIetfRateLimit', () => { + it('reads the key=value form', () => { + expect( + parseIetfRateLimit( + { ratelimit: 'limit=100, remaining=0, reset=12' }, + { now: NOW } + ) + ).toEqual({ + waitMs: 12_000, + retryAt: new Date(NOW + 12_000), + remaining: 0, + source: 'header', + }); + }); + + it('reads the structured-field form and names the policy', () => { + expect( + parseIetfRateLimit( + { ratelimit: '"default";r=0;t=12' }, + { now: NOW } + ) + ).toEqual({ + waitMs: 12_000, + retryAt: new Date(NOW + 12_000), + remaining: 0, + policy: 'default', + source: 'header', + }); + }); + + it('picks the exhausted policy of a list', () => { + const hint = parseIetfRateLimit( + { ratelimit: '"burst";r=5;t=1, "daily";r=0;t=3600' }, + { now: NOW } + ); + expect(hint).toMatchObject({ + waitMs: 3_600_000, + remaining: 0, + policy: 'daily', + }); + }); + + it('breaks a tie between exhausted policies with the longer wait', () => { + const hint = parseIetfRateLimit( + { ratelimit: '"a";r=0;t=5, "b";r=0;t=60' }, + { now: NOW } + ); + expect(hint).toMatchObject({ waitMs: 60_000, policy: 'b' }); + }); + + it('falls back to the RateLimit-Policy value when the limit names no policy', () => { + const hint = parseIetfRateLimit( + { + ratelimit: 'limit=100, remaining=0, reset=12', + 'ratelimit-policy': '100;w=60', + }, + { now: NOW } + ); + expect(hint.policy).toBe('100;w=60'); + }); + + it.each([ + ['no header', {}], + ['a header with no reset', { ratelimit: 'limit=100, remaining=0' }], + ['a structured item with no t', { ratelimit: '"default";r=0' }], + ['garbage', { ratelimit: 'not a rate limit' }], + [ + 'a reset that is not a number', + { ratelimit: 'limit=1, reset=soon' }, + ], + ['undefined headers', undefined], + ])('returns null for %s', (_label, headers) => { + expect(parseIetfRateLimit(headers, { now: NOW })).toBeNull(); + }); + }); + + describe('BUILT_IN_PARSERS', () => { + it('names the three parsers with one signature', () => { + expect(Object.keys(BUILT_IN_PARSERS)).toEqual([ + 'retryAfter', + 'resetHeaders', + 'ietf', + ]); + expect( + BUILT_IN_PARSERS.retryAfter( + { 'retry-after': '4' }, + { now: NOW } + ).waitMs + ).toBe(4_000); + expect( + BUILT_IN_PARSERS.resetHeaders( + { 'x-ratelimit-reset': '4' }, + { now: NOW } + ).waitMs + ).toBe(4_000); + expect( + BUILT_IN_PARSERS.ietf( + { ratelimit: '"d";r=0;t=4' }, + { now: NOW } + ).waitMs + ).toBe(4_000); + }); + }); +}); diff --git a/packages/core/modules/requester/rate-limit/policy.js b/packages/core/modules/requester/rate-limit/policy.js new file mode 100644 index 000000000..8635feb5c --- /dev/null +++ b/packages/core/modules/requester/rate-limit/policy.js @@ -0,0 +1,304 @@ +const { + BUILT_IN_PARSERS, + definedOnly, + hintFromRetryAt, + hintFromWait, +} = require('./parsers'); + +const DEFAULT_PARSERS = ['retryAfter', 'resetHeaders', 'ietf']; +const MIN_HINTED_WAIT_MS = 1_000; +const JITTER_MAX_MS = 1_000; +const JITTER_RATIO = 0.1; + +const REASONS = new Set([ + 'burst', + 'daily', + 'monthly', + 'concurrency', + 'unknown', +]); +const CLASSIFY_SOURCES = new Set(['header', 'body', 'static']); +const SCOPES = new Set(['credential', 'entity', 'app']); + +const normalizedPolicies = new WeakSet(); +const policiesByClass = new WeakMap(); + +const isObject = (value) => value !== null && typeof value === 'object'; +const isThenable = (value) => typeof value?.then === 'function'; + +function isPlainObject(value) { + if (!isObject(value) || Array.isArray(value)) return false; + const prototype = Object.getPrototypeOf(value); + return prototype === Object.prototype || prototype === null; +} + +function readParsers(raw, owner) { + if (!Array.isArray(raw.parsers)) return [...DEFAULT_PARSERS]; + for (const name of raw.parsers) { + if (!Object.hasOwn(BUILT_IN_PARSERS, name)) { + throw new TypeError( + `${owner}.rateLimit.parsers: unknown parser "${String( + name + )}"; known: ${Object.keys(BUILT_IN_PARSERS).join(', ')}` + ); + } + } + return [...raw.parsers]; +} + +function normalizePolicy(raw, owner) { + if (!isPlainObject(raw)) return undefined; + + const policy = { + scope: + SCOPES.has(raw.scope) || typeof raw.scope === 'function' + ? raw.scope + : 'entity', + parsers: readParsers(raw, owner), + }; + if (typeof raw.classify === 'function') { + policy.classify = (signal) => raw.classify(signal); + } + if (Number.isFinite(raw.minRetryAfterMs) && raw.minRetryAfterMs >= 0) { + policy.minRetryAfterMs = raw.minRetryAfterMs; + } + if (Number.isInteger(raw.maxConcurrency) && raw.maxConcurrency > 0) { + policy.maxConcurrency = raw.maxConcurrency; + } + if (Array.isArray(raw.windows)) policy.windows = raw.windows; + if (isPlainObject(raw.userHints)) policy.userHints = raw.userHints; + + normalizedPolicies.add(policy); + return policy; +} + +function readRateLimitPolicy(ctor) { + if (typeof ctor !== 'function') return undefined; + if (policiesByClass.has(ctor)) return policiesByClass.get(ctor); + const policy = normalizePolicy(ctor.rateLimit, ctor.name || 'Requester'); + policiesByClass.set(ctor, policy); + return policy; +} + +function asPolicy(value) { + if (!isObject(value)) return undefined; + if (normalizedPolicies.has(value)) return value; + return normalizePolicy(value, 'policy'); +} + +function computeScopeKey(policy, requester) { + const label = requester?._telemetryModuleLabel?.() ?? 'unknown'; + const scope = policy?.scope ?? 'entity'; + const delegate = requester?.delegate; + + if (scope === 'app') return `${label}:app`; + if (scope === 'entity') { + const id = delegate?.entity?.id; + return id === undefined || id === null + ? undefined + : `${label}:entity:${id}`; + } + if (scope === 'credential') { + const credential = delegate?.credential; + const id = + credential?.id ?? + (typeof credential === 'string' ? credential : undefined); + return id === undefined || id === null + ? undefined + : `${label}:credential:${id}`; + } + if (typeof scope === 'function') { + try { + const value = scope(requester); + return typeof value === 'string' || Number.isFinite(value) + ? `${label}:custom:${value}` + : undefined; + } catch { + return undefined; + } + } + return undefined; +} + +function toDate(value) { + if (value === undefined || value === null) return null; + const date = value instanceof Date ? value : new Date(value); + return Number.isNaN(date.getTime()) ? null : date; +} + +function normalizeClassified(raw, now) { + if (!isObject(raw)) return null; + const retryAt = toDate(raw.retryAt); + if (!REASONS.has(raw.reason) && !Number.isFinite(raw.waitMs) && !retryAt) { + return null; + } + + const extra = { + reason: REASONS.has(raw.reason) ? raw.reason : 'unknown', + policy: typeof raw.policy === 'string' ? raw.policy : undefined, + remaining: Number.isFinite(raw.remaining) ? raw.remaining : undefined, + source: CLASSIFY_SOURCES.has(raw.source) ? raw.source : 'body', + }; + + let hint = retryAt ? hintFromRetryAt(retryAt, now, extra) : null; + if (!hint && Number.isFinite(raw.waitMs)) { + hint = hintFromWait(raw.waitMs, now, extra); + } + return { + reason: extra.reason, + policy: extra.policy, + remaining: extra.remaining, + hint, + }; +} + +function runClassify(policy, signal, now, onClassifyError) { + if (typeof policy?.classify !== 'function') return null; + let raw; + try { + raw = policy.classify({ + status: signal.status, + headers: signal.headers, + body: signal.body, + }); + if (isThenable(raw)) { + Promise.resolve(raw).catch(() => {}); + throw new TypeError( + 'classify() must return a hint or null, not a Promise' + ); + } + } catch (error) { + if (!onClassifyError) throw error; + onClassifyError(error); + return null; + } + return normalizeClassified(raw, now); +} + +function parsedHint(policy, headers, now) { + for (const name of policy?.parsers ?? DEFAULT_PARSERS) { + const hint = BUILT_IN_PARSERS[name](headers, { now }); + if (hint) return hint; + } + return null; +} + +function staticHint(policy, reason, now) { + const window = policy?.windows?.find( + (candidate) => + candidate && + candidate.name === reason && + Number.isFinite(candidate.perMs) && + candidate.perMs > 0 + ); + const waitMs = Math.max(window?.perMs ?? 0, policy?.minRetryAfterMs ?? 0); + return waitMs > 0 ? hintFromWait(waitMs, now, { source: 'static' }) : null; +} + +function withClassified(hint, classified) { + return { + ...hint, + reason: classified?.reason ?? 'unknown', + ...definedOnly({ + policy: classified?.policy ?? hint.policy, + remaining: classified?.remaining ?? hint.remaining, + }), + }; +} + +function resolveCandidates(policy, signal, { now, onClassifyError }) { + const classified = runClassify(policy, signal, now, onClassifyError); + if (signal.status !== 429 && !classified) return { hint: null, classified }; + if (classified?.hint) return { hint: classified.hint, classified }; + + const parsed = parsedHint(policy, signal.headers, now); + if (parsed) return { hint: withClassified(parsed, classified), classified }; + + const fromPolicy = staticHint(policy, classified?.reason, now); + if (fromPolicy) { + return { hint: withClassified(fromPolicy, classified), classified }; + } + return { hint: null, classified }; +} + +function classifyRateLimit(policy, signal, options = {}) { + const { now = Date.now(), onClassifyError } = options; + return resolveCandidates(asPolicy(policy), signal, { now, onClassifyError }) + .hint; +} + +function resolveRateLimitHint(policy, signal, options = {}) { + const { + attempt = 0, + backOff = [], + now = Date.now(), + onClassifyError, + } = options; + const { hint, classified } = resolveCandidates(asPolicy(policy), signal, { + now, + onClassifyError, + }); + if (hint) return hint; + if (signal.status !== 429 && !classified) return null; + if (signal.status !== 429) { + return { + reason: classified.reason, + ...definedOnly({ + policy: classified.policy, + remaining: classified.remaining, + }), + }; + } + + const waitMs = + attempt < backOff.length ? Number(backOff[attempt]) * 1000 || 0 : 0; + return { + waitMs, + retryAt: new Date(now + waitMs), + reason: classified?.reason ?? 'unknown', + ...definedOnly({ + policy: classified?.policy, + remaining: classified?.remaining, + }), + source: 'backoff', + }; +} + +function computeWaitMs({ hint, policy, budgetMs, random = Math.random }) { + const base = Math.max( + hint.waitMs, + policy?.minRetryAfterMs ?? 0, + MIN_HINTED_WAIT_MS + ); + if (base > budgetMs) return { waitMs: base, fits: false }; + + const jitter = Math.floor( + random() * Math.min(JITTER_MAX_MS, base * JITTER_RATIO) + ); + return { waitMs: Math.min(base + jitter, budgetMs), fits: true }; +} + +function inProcessBudgetMs({ + maxInProcessWaitMs, + requestTimeoutMs = 0, + remainingMs = Infinity, + waitedMs = 0, +} = {}) { + const reserveMs = Math.min(requestTimeoutMs || 0, remainingMs / 2); + return Math.max( + 0, + Math.min(maxInProcessWaitMs - waitedMs, remainingMs - reserveMs) + ); +} + +module.exports = { + DEFAULT_PARSERS, + JITTER_MAX_MS, + MIN_HINTED_WAIT_MS, + classifyRateLimit, + computeScopeKey, + computeWaitMs, + inProcessBudgetMs, + readRateLimitPolicy, + resolveRateLimitHint, +}; diff --git a/packages/core/modules/requester/rate-limit/policy.test.js b/packages/core/modules/requester/rate-limit/policy.test.js new file mode 100644 index 000000000..2ce5ddf76 --- /dev/null +++ b/packages/core/modules/requester/rate-limit/policy.test.js @@ -0,0 +1,921 @@ +const { + DEFAULT_PARSERS, + JITTER_MAX_MS, + MIN_HINTED_WAIT_MS, + classifyRateLimit, + computeScopeKey, + computeWaitMs, + inProcessBudgetMs, + readRateLimitPolicy, + resolveRateLimitHint, +} = require('./policy'); + +const NOW = Date.parse('2026-09-28T12:00:00.000Z'); +const options = { now: NOW }; + +describe('rate-limit/policy', () => { + it('exposes the defaults', () => { + expect(DEFAULT_PARSERS).toEqual(['retryAfter', 'resetHeaders', 'ietf']); + expect(MIN_HINTED_WAIT_MS).toBe(1_000); + expect(JITTER_MAX_MS).toBe(1_000); + }); + + describe('readRateLimitPolicy', () => { + it('returns undefined when the class declares nothing', () => { + class Api {} + expect(readRateLimitPolicy(Api)).toBeUndefined(); + }); + + it.each([ + ['a number', 5], + ['a string', 'fast'], + ['an array', []], + ['null', null], + ])('returns undefined when rateLimit is %s', (_label, value) => { + class Api { + static rateLimit = value; + } + expect(readRateLimitPolicy(Api)).toBeUndefined(); + }); + + it('returns undefined for something that is not a class', () => { + expect(readRateLimitPolicy(undefined)).toBeUndefined(); + expect(readRateLimitPolicy({ rateLimit: {} })).toBeUndefined(); + }); + + it('defaults the scope to entity and the parsers to the built-ins', () => { + class Api { + static rateLimit = {}; + } + expect(readRateLimitPolicy(Api)).toEqual({ + scope: 'entity', + parsers: DEFAULT_PARSERS, + }); + }); + + it('keeps the known keys and ignores the unknown ones', () => { + const windows = [{ name: 'burst', limit: 100, perMs: 10_000 }]; + const userHints = { daily: { links: [] } }; + class Api { + static rateLimit = { + scope: 'credential', + windows, + maxConcurrency: 5, + minRetryAfterMs: 60_000, + maxInProcessWaitMs: 30_000, + parsers: ['retryAfter'], + userHints, + somethingElse: true, + }; + } + const policy = readRateLimitPolicy(Api); + expect(policy).toMatchObject({ + scope: 'credential', + windows, + maxConcurrency: 5, + minRetryAfterMs: 60_000, + parsers: ['retryAfter'], + userHints, + }); + expect(policy).not.toHaveProperty('somethingElse'); + expect(policy).not.toHaveProperty('maxInProcessWaitMs'); + }); + + it('drops values of the wrong type', () => { + class Api { + static rateLimit = { + classify: 'nope', + minRetryAfterMs: -1, + maxConcurrency: 1.5, + windows: 'many', + userHints: 3, + scope: 'galaxy', + }; + } + expect(readRateLimitPolicy(Api)).toEqual({ + scope: 'entity', + parsers: DEFAULT_PARSERS, + }); + }); + + it('falls back to the default parsers when parsers is not an array', () => { + class Api { + static rateLimit = { parsers: 'retryAfter' }; + } + expect(readRateLimitPolicy(Api).parsers).toEqual(DEFAULT_PARSERS); + }); + + it('accepts an empty parser list', () => { + class Api { + static rateLimit = { parsers: [] }; + } + expect(readRateLimitPolicy(Api).parsers).toEqual([]); + }); + + it('accepts a function as the scope', () => { + const scope = () => 'x'; + class Api { + static rateLimit = { scope }; + } + expect(readRateLimitPolicy(Api).scope).toBe(scope); + }); + + it('throws a TypeError that names the class and the known parsers', () => { + class HubApi { + static rateLimit = { parsers: ['retryAfter', 'magic'] }; + } + expect(() => readRateLimitPolicy(HubApi)).toThrow(TypeError); + expect(() => readRateLimitPolicy(HubApi)).toThrow( + 'HubApi.rateLimit.parsers: unknown parser "magic"; known: retryAfter, resetHeaders, ietf' + ); + }); + + it('calls classify as a method of the declared object', () => { + class Api { + static rateLimit = { + reason: 'burst', + classify() { + return { reason: this.reason, waitMs: 1_000 }; + }, + }; + } + const policy = readRateLimitPolicy(Api); + expect(policy.classify({ status: 429 })).toEqual({ + reason: 'burst', + waitMs: 1_000, + }); + }); + + it('is memoized per class and resolved through subclasses', () => { + class Base { + static rateLimit = { minRetryAfterMs: 5_000 }; + } + class Child extends Base {} + const first = readRateLimitPolicy(Child); + expect(readRateLimitPolicy(Child)).toBe(first); + expect(first.minRetryAfterMs).toBe(5_000); + }); + }); + + describe('computeScopeKey', () => { + const requester = (delegate) => ({ + _telemetryModuleLabel: () => 'hubspot', + delegate, + }); + + it('keys an entity scope on the entity id, by default', () => { + const subject = requester({ entity: { id: 7 } }); + expect(computeScopeKey(undefined, subject)).toBe( + 'hubspot:entity:7' + ); + expect(computeScopeKey({ scope: 'entity' }, subject)).toBe( + 'hubspot:entity:7' + ); + }); + + it('keys a credential scope on the credential id or a string credential', () => { + expect( + computeScopeKey( + { scope: 'credential' }, + requester({ credential: { id: 3 } }) + ) + ).toBe('hubspot:credential:3'); + expect( + computeScopeKey( + { scope: 'credential' }, + requester({ credential: 'c-9' }) + ) + ).toBe('hubspot:credential:c-9'); + }); + + it('keys an app scope on the module alone', () => { + expect(computeScopeKey({ scope: 'app' }, requester({}))).toBe( + 'hubspot:app' + ); + }); + + it('keys a function scope on its result', () => { + const scope = (subject) => subject.delegate.entity.externalId; + expect( + computeScopeKey( + { scope }, + requester({ entity: { externalId: 'portal-1' } }) + ) + ).toBe('hubspot:custom:portal-1'); + }); + + it.each([ + ['no entity', { scope: 'entity' }, requester({})], + ['no credential', { scope: 'credential' }, requester({})], + ['no delegate', { scope: 'entity' }, requester(undefined)], + [ + 'a function that throws', + { + scope: () => { + throw new Error('boom'); + }, + }, + requester({}), + ], + [ + 'a function that returns nothing', + { scope: () => undefined }, + requester({}), + ], + [ + 'an async function', + { scope: async () => 'portal-1' }, + requester({}), + ], + [ + 'a function that returns an object', + { scope: () => ({ id: 'portal-1' }) }, + requester({}), + ], + ])('returns undefined for %s', (_label, policy, subject) => { + expect(computeScopeKey(policy, subject)).toBeUndefined(); + }); + }); + + describe('classifyRateLimit', () => { + it('returns null with no policy and no headers', () => { + expect( + classifyRateLimit( + undefined, + { status: 429, headers: {} }, + options + ) + ).toBeNull(); + }); + + it('reads Retry-After on a 429 with no policy', () => { + expect( + classifyRateLimit( + undefined, + { status: 429, headers: { 'retry-after': '5' } }, + options + ) + ).toEqual({ + waitMs: 5_000, + retryAt: new Date(NOW + 5_000), + reason: 'unknown', + source: 'header', + }); + }); + + it('does not read headers on a status other than 429 that nothing classified', () => { + expect( + classifyRateLimit( + undefined, + { status: 503, headers: { 'retry-after': '5' } }, + options + ) + ).toBeNull(); + expect( + classifyRateLimit( + { classify: () => null }, + { status: 403, headers: { 'retry-after': '5' } }, + options + ) + ).toBeNull(); + }); + + it('takes a classify hint that carries a time, and defaults its source to body', () => { + const policy = { + classify: () => ({ reason: 'daily', waitMs: 60_000 }), + }; + expect( + classifyRateLimit(policy, { status: 403, headers: {} }, options) + ).toEqual({ + waitMs: 60_000, + retryAt: new Date(NOW + 60_000), + reason: 'daily', + source: 'body', + }); + }); + + it('keeps a classify retryAt, as a Date, a timestamp or a number', () => { + const retryAt = new Date(NOW + 3_600_000); + for (const value of [ + retryAt, + retryAt.toISOString(), + retryAt.getTime(), + ]) { + const policy = { + classify: () => ({ reason: 'daily', retryAt: value }), + }; + const hint = classifyRateLimit( + policy, + { status: 429, headers: {} }, + options + ); + expect(hint.waitMs).toBe(3_600_000); + expect(hint.retryAt).toEqual(retryAt); + } + }); + + it('keeps the source classify gives, when it is a known one', () => { + const build = (source) => ({ + classify: () => ({ waitMs: 1_000, source }), + }); + const signal = { status: 429, headers: {} }; + expect( + classifyRateLimit(build('static'), signal, options).source + ).toBe('static'); + expect( + classifyRateLimit(build('backoff'), signal, options).source + ).toBe('body'); + }); + + it('lets a classify time win over Retry-After', () => { + const policy = { + classify: () => ({ reason: 'daily', waitMs: 60_000 }), + }; + const hint = classifyRateLimit( + policy, + { status: 429, headers: { 'retry-after': '2' } }, + options + ); + expect(hint.waitMs).toBe(60_000); + }); + + it('takes the time of a reason-only classify from the first parser', () => { + const policy = { + classify: () => ({ reason: 'burst', policy: 'SECONDLY' }), + }; + expect( + classifyRateLimit( + policy, + { status: 429, headers: { 'retry-after': '7' } }, + options + ) + ).toEqual({ + waitMs: 7_000, + retryAt: new Date(NOW + 7_000), + reason: 'burst', + policy: 'SECONDLY', + source: 'header', + }); + }); + + it('takes the time of a reason-only classify from the window when no header helps', () => { + const policy = { + windows: [{ name: 'burst', limit: 110, perMs: 10_000 }], + classify: () => ({ reason: 'burst' }), + }; + expect( + classifyRateLimit(policy, { status: 429, headers: {} }, options) + ).toEqual({ + waitMs: 10_000, + retryAt: new Date(NOW + 10_000), + reason: 'burst', + source: 'static', + }); + }); + + it('returns null for a reason-only classify with no time anywhere', () => { + const policy = { classify: () => ({ reason: 'daily' }) }; + expect( + classifyRateLimit(policy, { status: 403, headers: {} }, options) + ).toBeNull(); + }); + + it('treats a classify time that is not valid as no time', () => { + const policy = { + classify: () => ({ reason: 'daily', waitMs: -5 }), + }; + expect( + classifyRateLimit(policy, { status: 403, headers: {} }, options) + ).toBeNull(); + }); + + it('turns a static minRetryAfterMs into the hint of a bare 429', () => { + expect( + classifyRateLimit( + { minRetryAfterMs: 60_000 }, + { status: 429, headers: {} }, + options + ) + ).toEqual({ + waitMs: 60_000, + retryAt: new Date(NOW + 60_000), + reason: 'unknown', + source: 'static', + }); + }); + + it('reads the parsers in the order the policy lists them', () => { + const headers = { 'retry-after': '5', 'x-ratelimit-reset': '30' }; + const signal = { status: 429, headers }; + expect( + classifyRateLimit( + { parsers: ['resetHeaders', 'retryAfter'] }, + signal, + options + ).waitMs + ).toBe(30_000); + expect(classifyRateLimit(undefined, signal, options).waitMs).toBe( + 5_000 + ); + }); + + it('does not use a parser the policy leaves out', () => { + expect( + classifyRateLimit( + { parsers: ['retryAfter'] }, + { status: 429, headers: { 'x-ratelimit-reset': '30' } }, + options + ) + ).toBeNull(); + }); + + it('carries remaining and policy from a parser', () => { + const hint = classifyRateLimit( + undefined, + { + status: 429, + headers: { ratelimit: '"daily";r=0;t=3600' }, + }, + options + ); + expect(hint).toMatchObject({ + remaining: 0, + policy: 'daily', + waitMs: 3_600_000, + }); + }); + + it('turns a reason it does not know into unknown', () => { + const policy = { + classify: () => ({ reason: 'hourly', waitMs: 1_000 }), + }; + expect( + classifyRateLimit(policy, { status: 429, headers: {} }, options) + .reason + ).toBe('unknown'); + }); + + it('hands classify the status, the headers and the body', () => { + const classify = jest.fn(() => null); + const headers = { 'x-a': '1' }; + const body = { code: 'LIMIT' }; + classifyRateLimit( + { classify }, + { status: 403, headers, body }, + options + ); + expect(classify).toHaveBeenCalledWith({ + status: 403, + headers, + body, + }); + }); + + it('lets a throwing classify propagate when nobody listens', () => { + const policy = { + classify: () => { + throw new Error('bad classifier'); + }, + }; + expect(() => + classifyRateLimit(policy, { status: 429, headers: {} }, options) + ).toThrow('bad classifier'); + }); + + it('reports a throwing classify and goes on with the parsers', () => { + const error = new Error('bad classifier'); + const policy = { + classify: () => { + throw error; + }, + }; + const onClassifyError = jest.fn(); + const hint = classifyRateLimit( + policy, + { status: 429, headers: { 'retry-after': '4' } }, + { ...options, onClassifyError } + ); + expect(onClassifyError).toHaveBeenCalledWith(error); + expect(hint.waitMs).toBe(4_000); + }); + + it.each([ + ['an empty object', {}], + ['an undefined reason', { reason: undefined }], + ['a reason it does not know', { reason: 'hourly' }], + ['only a policy name', { policy: 'DAILY' }], + ['a retryAt that is not a date', { retryAt: 'soon' }], + ['a waitMs that is not a number', { waitMs: '5000' }], + ])('treats %s from classify as no limit', (_label, result) => { + const policy = { classify: () => result }; + expect( + classifyRateLimit( + policy, + { status: 404, headers: { 'retry-after': '5' } }, + options + ) + ).toBeNull(); + expect( + resolveRateLimitHint( + policy, + { status: 404, headers: {} }, + { ...options, attempt: 0, backOff: [1] } + ) + ).toBeNull(); + }); + + it.each([ + ['a finite waitMs', { waitMs: 2_000 }, 2_000], + ['a valid retryAt', { retryAt: new Date(NOW + 3_000) }, 3_000], + ])( + 'counts a classify result with %s and no reason as a limit', + (_label, result, waitMs) => { + expect( + classifyRateLimit( + { classify: () => result }, + { status: 404, headers: {} }, + options + ) + ).toMatchObject({ waitMs, reason: 'unknown', source: 'body' }); + } + ); + + it('reports a classify that returns a Promise and goes on with the parsers', () => { + const onClassifyError = jest.fn(); + const policy = { + classify: async () => ({ reason: 'daily', waitMs: 60_000 }), + }; + const hint = classifyRateLimit( + policy, + { status: 429, headers: { 'retry-after': '4' } }, + { ...options, onClassifyError } + ); + expect(onClassifyError).toHaveBeenCalledWith( + expect.objectContaining({ + name: 'TypeError', + message: + 'classify() must return a hint or null, not a Promise', + }) + ); + expect(hint).toMatchObject({ + waitMs: 4_000, + reason: 'unknown', + source: 'header', + }); + }); + + it('treats a classify that returns a Promise as no limit for a status other than 429', () => { + const policy = { classify: async () => ({ reason: 'daily' }) }; + const onClassifyError = jest.fn(); + expect( + classifyRateLimit( + policy, + { status: 404, headers: {} }, + { ...options, onClassifyError } + ) + ).toBeNull(); + expect(onClassifyError).toHaveBeenCalledTimes(1); + }); + + it('lets the Promise error propagate when nobody listens', () => { + expect(() => + classifyRateLimit( + { classify: async () => null }, + { status: 429, headers: {} }, + options + ) + ).toThrow(TypeError); + }); + + it('handles the rejection of a Promise that classify returns', async () => { + const then = jest.fn(); + classifyRateLimit( + { classify: () => ({ then }) }, + { status: 429, headers: {} }, + { ...options, onClassifyError: () => {} } + ); + await Promise.resolve(); + expect(then).toHaveBeenCalledWith( + expect.any(Function), + expect.any(Function) + ); + }); + + it('accepts the policy object a module declares, without reading it first', () => { + const declared = { + parsers: ['retryAfter'], + classify: () => ({ reason: 'daily', waitMs: 1_000 }), + }; + expect( + classifyRateLimit( + declared, + { status: undefined, headers: {} }, + options + ).reason + ).toBe('daily'); + }); + }); + + describe('resolveRateLimitHint', () => { + const backOff = [1, 3, 10]; + + it('returns the hint when there is one', () => { + const hint = resolveRateLimitHint( + undefined, + { status: 429, headers: { 'retry-after': '9' } }, + { ...options, attempt: 0, backOff } + ); + expect(hint).toMatchObject({ waitMs: 9_000, source: 'header' }); + }); + + it('falls back to the ladder step with source backoff', () => { + const hint = resolveRateLimitHint( + undefined, + { status: 429, headers: {} }, + { ...options, attempt: 2, backOff } + ); + expect(hint).toEqual({ + waitMs: 10_000, + retryAt: new Date(NOW + 10_000), + reason: 'unknown', + source: 'backoff', + }); + }); + + it('returns null for a status other than 429 that classify did not recognise', () => { + expect( + resolveRateLimitHint( + { classify: () => null }, + { status: 403, headers: { 'retry-after': '5' } }, + { ...options, attempt: 0, backOff } + ) + ).toBeNull(); + expect( + resolveRateLimitHint( + undefined, + { status: 503, headers: {} }, + { ...options, attempt: 0, backOff } + ) + ).toBeNull(); + }); + + it('returns an untimed hint for a status other than 429 that classify named by reason only', () => { + const policy = { + classify: () => ({ + reason: 'daily', + policy: 'DAILY', + remaining: 0, + }), + }; + const hint = resolveRateLimitHint( + policy, + { status: 403, headers: {} }, + { ...options, attempt: 1, backOff } + ); + expect(hint).toEqual({ + reason: 'daily', + policy: 'DAILY', + remaining: 0, + }); + expect(hint).not.toHaveProperty('waitMs'); + expect(hint).not.toHaveProperty('retryAt'); + expect(hint).not.toHaveProperty('source'); + }); + + it('keeps classifyRateLimit null for a reason-only classify on a status other than 429', () => { + const policy = { classify: () => ({ reason: 'daily' }) }; + expect( + classifyRateLimit(policy, { status: 403, headers: {} }, options) + ).toBeNull(); + }); + + it('waits 0 ms past the end of the ladder', () => { + const hint = resolveRateLimitHint( + undefined, + { status: 429, headers: {} }, + { ...options, attempt: 3, backOff } + ); + expect(hint).toMatchObject({ waitMs: 0, source: 'backoff' }); + }); + + it('keeps the reason and policy of a reason-only classify in the fallback', () => { + const policy = { + classify: () => ({ reason: 'daily', policy: 'DAILY' }), + }; + const hint = resolveRateLimitHint( + policy, + { status: 429, headers: {} }, + { ...options, attempt: 0, backOff } + ); + expect(hint).toMatchObject({ + reason: 'daily', + policy: 'DAILY', + source: 'backoff', + waitMs: 1_000, + }); + }); + + it('reports a throwing classify and still answers', () => { + const onClassifyError = jest.fn(); + const policy = { + classify: () => { + throw new Error('boom'); + }, + }; + const hint = resolveRateLimitHint( + policy, + { status: 429, headers: {} }, + { ...options, attempt: 0, backOff, onClassifyError } + ); + expect(onClassifyError).toHaveBeenCalledTimes(1); + expect(hint.source).toBe('backoff'); + }); + }); + + describe('computeWaitMs', () => { + const hintOf = (waitMs) => ({ + waitMs, + retryAt: new Date(NOW + waitMs), + source: 'header', + }); + const noJitter = () => 0; + const maxJitter = () => 0.999999; + + it('waits as long as the hint says', () => { + expect( + computeWaitMs({ + hint: hintOf(5_000), + budgetMs: 300_000, + random: noJitter, + }) + ).toEqual({ waitMs: 5_000, fits: true }); + }); + + it('applies minRetryAfterMs as a floor', () => { + expect( + computeWaitMs({ + hint: hintOf(5_000), + policy: { minRetryAfterMs: 60_000 }, + budgetMs: 300_000, + random: noJitter, + }).waitMs + ).toBe(60_000); + }); + + it('applies the minimum wait', () => { + expect( + computeWaitMs({ + hint: hintOf(0), + budgetMs: 300_000, + random: noJitter, + }).waitMs + ).toBe(MIN_HINTED_WAIT_MS); + }); + + it('adds jitter of at most 10 percent of the wait, and at most 1 s', () => { + expect( + computeWaitMs({ + hint: hintOf(5_000), + budgetMs: 300_000, + random: maxJitter, + }).waitMs + ).toBe(5_499); + expect( + computeWaitMs({ + hint: hintOf(60_000), + budgetMs: 300_000, + random: maxJitter, + }).waitMs + ).toBe(60_999); + }); + + it('never lets jitter push a wait over the budget', () => { + expect( + computeWaitMs({ + hint: hintOf(5_000), + budgetMs: 5_100, + random: maxJitter, + }) + ).toEqual({ waitMs: 5_100, fits: true }); + }); + + it('fits a wait that equals the budget', () => { + expect( + computeWaitMs({ + hint: hintOf(5_000), + budgetMs: 5_000, + random: maxJitter, + }) + ).toEqual({ waitMs: 5_000, fits: true }); + }); + + it('reports that a wait over the budget does not fit, with no jitter', () => { + expect( + computeWaitMs({ + hint: hintOf(5_000), + budgetMs: 4_000, + random: maxJitter, + }) + ).toEqual({ waitMs: 5_000, fits: false }); + }); + + it('uses Math.random when no source of randomness is given', () => { + const { waitMs } = computeWaitMs({ + hint: hintOf(5_000), + budgetMs: 300_000, + }); + expect(waitMs).toBeGreaterThanOrEqual(5_000); + expect(waitMs).toBeLessThan(5_500); + }); + }); + + describe('inProcessBudgetMs', () => { + it('takes the cap from the caller', () => { + expect( + inProcessBudgetMs({ + maxInProcessWaitMs: 30_000, + remainingMs: Infinity, + }) + ).toBe(30_000); + expect( + inProcessBudgetMs({ + maxInProcessWaitMs: 0, + remainingMs: Infinity, + }) + ).toBe(0); + }); + + it('subtracts the wait already spent', () => { + expect( + inProcessBudgetMs({ + maxInProcessWaitMs: 300_000, + remainingMs: Infinity, + waitedMs: 100_000, + }) + ).toBe(200_000); + }); + + it('keeps one request timeout of the time left for the next request', () => { + expect( + inProcessBudgetMs({ + maxInProcessWaitMs: 300_000, + requestTimeoutMs: 30_000, + remainingMs: 100_000, + }) + ).toBe(70_000); + }); + + it('keeps at most half of the time left for the next request', () => { + expect( + inProcessBudgetMs({ + maxInProcessWaitMs: 300_000, + requestTimeoutMs: 60_000, + remainingMs: 29_000, + }) + ).toBe(14_500); + }); + + it('is the cap when the invocation has time to spare', () => { + expect( + inProcessBudgetMs({ + maxInProcessWaitMs: 300_000, + requestTimeoutMs: 60_000, + remainingMs: 900_000, + }) + ).toBe(300_000); + }); + + it('is never negative', () => { + expect( + inProcessBudgetMs({ + maxInProcessWaitMs: 300_000, + requestTimeoutMs: 60_000, + remainingMs: 0, + }) + ).toBe(0); + expect( + inProcessBudgetMs({ + maxInProcessWaitMs: 300_000, + remainingMs: Infinity, + waitedMs: 400_000, + }) + ).toBe(0); + }); + + it('treats a missing request timeout as 0', () => { + expect( + inProcessBudgetMs({ + maxInProcessWaitMs: 300_000, + remainingMs: 50_000, + }) + ).toBe(50_000); + expect( + inProcessBudgetMs({ + maxInProcessWaitMs: 300_000, + requestTimeoutMs: null, + remainingMs: 50_000, + }) + ).toBe(50_000); + }); + }); +}); diff --git a/packages/core/modules/requester/rate-limit/public-api.test.js b/packages/core/modules/requester/rate-limit/public-api.test.js new file mode 100644 index 000000000..749627afd --- /dev/null +++ b/packages/core/modules/requester/rate-limit/public-api.test.js @@ -0,0 +1,26 @@ +const core = require('../../../index'); +const { RateLimitError } = require('../../../errors'); +const rateLimit = require('./index'); +const deadline = require('../../../core/invocation-deadline'); + +describe('rate-limit public API', () => { + it('exports the error class from the package root', () => { + expect(core.RateLimitError).toBe(RateLimitError); + }); + + it.each([ + 'classifyRateLimit', + 'parseRetryAfter', + 'parseResetHeaders', + 'parseIetfRateLimit', + ])('exports %s from the package root', (name) => { + expect(core[name]).toBe(rateLimit[name]); + }); + + it.each(['runWithInvocationDeadline', 'remainingInvocationMs'])( + 'exports %s from the package root', + (name) => { + expect(core[name]).toBe(deadline[name]); + } + ); +}); diff --git a/packages/core/modules/requester/requester.js b/packages/core/modules/requester/requester.js index 514256c15..d81d39a17 100644 --- a/packages/core/modules/requester/requester.js +++ b/packages/core/modules/requester/requester.js @@ -1,13 +1,24 @@ const fetch = require('node-fetch'); const { AsyncLocalStorage } = require('async_hooks'); const { Delegate } = require('../../core'); -const { FetchError } = require('../../errors'); +const { + invocationMaxInProcessWaitMs, + remainingInvocationMs, +} = require('../../core/invocation-deadline'); +const { FetchError, RateLimitError } = require('../../errors'); const { get } = require('../../assertions'); const { getTelemetry } = require('../../telemetry/telemetry-runtime'); const { getLogger } = require('../../logs'); const { redactUrl } = require('../../logs/redact'); const { toSanitizedSurrogate } = require('../../logs/serialize'); const { getLoggerScope } = require('../../logs/context'); +const { + computeScopeKey, + computeWaitMs, + inProcessBudgetMs, + readRateLimitPolicy, + resolveRateLimitHint, +} = require('./rate-limit'); const DEFAULT_REQUEST_TIMEOUT_MS = 60_000; @@ -22,6 +33,10 @@ function sanitizedCause(err) { } const MAX_AUTH_RETRIES = 3; +// Hinted waits are bounded by time, but a provider that always answers with a +// short Retry-After would still keep the request in a long loop. +const MAX_HINTED_RETRIES = 10; + // This context marks the async call chain that holds the refresh slot, and // names the requester that holds it. Token requests re-enter _rawRequest // through this._post. A 401 from inside that chain must fail fast: if it @@ -34,6 +49,8 @@ class Requester extends Delegate { constructor(params) { super(params); this.backOff = get(params, 'backOff', [1, 3, 10, 30, 60, 180]); + this._rateLimitPolicy = readRateLimitPolicy(this.constructor); + this._random = params?.random ?? Math.random; this.isRefreshable = false; this.refreshCount = 0; this.authGraceRetryCount = 0; @@ -211,9 +228,16 @@ class Requester extends Delegate { * @param {Object} options - Fetch options, as built by `_request`. * @param {number} attempt - 0-based count of retries already made for * this call. Indexes `this.backOff` for the next delay and is passed - * back in on each recursive retry. + * back in on each recursive retry. A hinted retry leaves it unchanged. + * @param {{ waitedMs: number, retries: number }} hinted - The time slept + * and the retries made for rate-limit hints, for the whole call. */ - async _rawRequest(url, options, attempt = 0) { + async _rawRequest( + url, + options, + attempt = 0, + hinted = { waitedMs: 0, retries: 0 } + ) { let encodedUrl = encodeURI(url); if (options.query) { let queryBuild = '?'; @@ -279,7 +303,7 @@ class Requester extends Delegate { clearRequestTimer(); const delay = this.backOff[attempt] * 1000; await new Promise((resolve) => setTimeout(resolve, delay)); - return this._rawRequest(url, options, attempt + 1); + return this._rawRequest(url, options, attempt + 1, hinted); } const fetchError = await FetchError.create({ resource: encodedUrl, @@ -298,15 +322,41 @@ class Requester extends Delegate { const { status } = response; + const throttle = await this._detectThrottle( + response, + status, + attempt + ); + const hintedDelayMs = await this._hintedRetryDelayMs({ + throttle, + status, + attempt, + hinted, + encodedUrl, + options, + response, + timeoutMs, + }); + if (hintedDelayMs !== null) { + clearRequestTimer(); + await new Promise((resolve) => + setTimeout(resolve, hintedDelayMs) + ); + return this._rawRequest(url, options, attempt, { + waitedMs: hinted.waitedMs + hintedDelayMs, + retries: hinted.retries + 1, + }); + } + // If the status is retriable and there are back off requests left, retry the request if ( - (status === 429 || status >= 500) && + (throttle?.throttled || status >= 500) && attempt < this.backOff.length ) { clearRequestTimer(); const delay = this.backOff[attempt] * 1000; await new Promise((resolve) => setTimeout(resolve, delay)); - return this._rawRequest(url, options, attempt + 1); + return this._rawRequest(url, options, attempt + 1, hinted); } if (status === 401) { @@ -327,7 +377,7 @@ class Requester extends Delegate { // This request did not try the current token. Retry with // it. Do not spend one more provider-side rotation. clearRequestTimer(); - return this._rawRequest(url, options, attempt + 1); + return this._rawRequest(url, options, attempt + 1, hinted); } if (!this.isRefreshable) { @@ -344,7 +394,12 @@ class Requester extends Delegate { await new Promise((resolve) => setTimeout(resolve, delay) ); - return this._rawRequest(url, options, attempt + 1); + return this._rawRequest( + url, + options, + attempt + 1, + hinted + ); } throw await this._invalidateAuth( @@ -377,7 +432,7 @@ class Requester extends Delegate { const refreshSucceeded = await this._refreshAuthOnce(); if (refreshSucceeded) { clearRequestTimer(); - return this._rawRequest(url, options, attempt + 1); + return this._rawRequest(url, options, attempt + 1, hinted); } throw await this._invalidateAuth(encodedUrl, options, response); @@ -392,6 +447,7 @@ class Requester extends Delegate { resource: encodedUrl, init: options, response, + responseBody: throttle?.responseBody, }); throw this._maybeFlagTimeoutDuringBodyRead( fetchError, @@ -421,6 +477,128 @@ class Requester extends Delegate { } } + async _hintedRetryDelayMs({ + throttle, + status, + attempt, + hinted, + encodedUrl, + options, + response, + timeoutMs, + }) { + if (!throttle?.throttled || throttle.hint.source === 'backoff') { + return null; + } + const { hint } = throttle; + const untimed = hint.waitMs === undefined; + const mayWait = + this.backOff.length > 0 && hinted.retries < MAX_HINTED_RETRIES; + const budgetMs = mayWait + ? inProcessBudgetMs({ + maxInProcessWaitMs: invocationMaxInProcessWaitMs(), + requestTimeoutMs: this.requestTimeoutMs, + remainingMs: remainingInvocationMs(), + waitedMs: hinted.waitedMs, + }) + : 0; + const { waitMs, fits } = untimed + ? { waitMs: undefined, fits: false } + : computeWaitMs({ + hint, + policy: this._rateLimitPolicy, + budgetMs, + random: this._random, + }); + if (fits) { + this._logRateLimited({ + status, + hint, + waitMs, + attempt, + waitedMs: hinted.waitedMs, + }); + return waitMs; + } + + this._logRequestFailed(encodedUrl, options, status); + const rateLimitError = await RateLimitError.create({ + resource: encodedUrl, + init: options, + response, + responseBody: throttle.responseBody, + hint, + waitMs, + module: this._telemetryModuleLabel(), + scopeKey: computeScopeKey(this._rateLimitPolicy, this), + }); + throw this._maybeFlagTimeoutDuringBodyRead(rateLimitError, timeoutMs); + } + + async _detectThrottle(response, status, attempt) { + const policy = this._rateLimitPolicy; + const canClassify = + typeof policy?.classify === 'function' && + status >= 400 && + status !== 401; + if (status !== 429 && !canClassify) return null; + + let responseBody; + let body; + if (canClassify) { + ({ text: responseBody, json: body } = + await this._readBodyForClassify(response)); + } + + const hint = resolveRateLimitHint( + policy, + { status, headers: response.headers, body }, + { + attempt, + backOff: this.backOff, + now: Date.now(), + onClassifyError: (error) => + this._logClassifyFailed(status, error), + } + ); + return { throttled: Boolean(hint), hint, responseBody }; + } + + async _readBodyForClassify(response) { + if (response.bodyUsed || typeof response.text !== 'function') { + return {}; + } + const text = await response.text(); + try { + return { text, json: JSON.parse(text) }; + } catch { + return { text }; + } + } + + _logRateLimited({ status, hint, waitMs, attempt, waitedMs }) { + const logger = this.logger; + logger.trace('Rate limited', { + eventName: `${logger.name}.rate_limited`, + statusCode: status, + waitMs, + retryAt: new Date(Date.now() + waitMs).toISOString(), + reason: hint.reason, + hintSource: hint.source, + attempt, + waitedMs, + }); + } + + _logClassifyFailed(status, error) { + const logger = this.logger; + logger.warn('Rate limit classifier failed', { + eventName: `${logger.name}.rate_limit_classify_failed`, + statusCode: status, + error, + }); + } + _logRequestFailed(encodedUrl, options, status) { const logger = this.logger; if (!logger.isLevelEnabled('DEBUG')) return; diff --git a/packages/core/modules/requester/requester.test.js b/packages/core/modules/requester/requester.test.js index a2203fa1a..bac68eb05 100644 --- a/packages/core/modules/requester/requester.test.js +++ b/packages/core/modules/requester/requester.test.js @@ -1,4 +1,8 @@ +const { Headers } = require('node-fetch'); const { Requester } = require('./requester'); +const { FetchError, RateLimitError } = require('../../errors'); +const { createMemorySink } = require('../../logs'); +const { runWithInvocationDeadline } = require('../../core/invocation-deadline'); /** * Requester is abstract: subclasses provide `addAuthHeaders`. For the @@ -39,30 +43,905 @@ function okFetch(body) { } describe('Requester', () => { - describe('429 and 5xx testing', () => { - let backOffArray = [1, 1, 1]; - let requester = new TestRequester({ backOff: backOffArray }); - let sum = backOffArray.reduce((a, b) => { - return a + b; - }, 0); - it.skip("should retry with 'exponential' back off due to 429", async () => { - let startTime = await Date.now(); - let res = await requester._get({ - url: 'https://70e18ff0-1967-4fb5-8f96-10477ab6bb9e.mock.pstmn.io//429', - }); - let endTime = await Date.now(); - let difference = endTime - startTime; - expect(difference).toBeGreaterThan(sum * 1000); - }); - - it.skip("should retry with 'exponential' back off due to 500", async () => { - let startTime = await Date.now(); - let res = await requester._get({ - url: 'https://70e18ff0-1967-4fb5-8f96-10477ab6bb9e.mock.pstmn.io//5xx', - }); - let endTime = await Date.now(); - let difference = endTime - startTime; - expect(difference).toBeGreaterThan(sum * 1000); + describe('rate-limit handling', () => { + const url = 'https://example.com/limited'; + const LADDER_MS = [1_000, 3_000, 10_000, 30_000, 60_000, 180_000]; + + function throttledFetch(responses) { + const queue = [...responses]; + const built = []; + const fetch = jest.fn(async () => { + const next = queue.shift(); + if (!next) throw new Error('unexpected fetch call'); + if (next.reject) throw next.reject; + const contentType = next.contentType ?? 'application/json'; + const text = next.text ?? JSON.stringify(next.body ?? {}); + const response = { + status: next.status, + bodyUsed: false, + headers: + next.headersObject ?? + new Map( + Object.entries({ + 'Content-Type': contentType, + ...(next.headers ?? {}), + }) + ), + json: async () => next.body ?? {}, + text: jest.fn(async () => text), + }; + built.push(response); + return response; + }); + fetch.responses = built; + return fetch; + } + + const limited = (headers, extra = {}) => ({ + status: 429, + headers, + ...extra, + }); + const ok = { status: 200, body: { ok: true } }; + + let delays; + let sink; + + beforeEach(() => { + delays = []; + const realSetTimeout = global.setTimeout; + jest.spyOn(global, 'setTimeout').mockImplementation( + (fn, delay, ...args) => { + delays.push(delay); + return realSetTimeout(fn, 0, ...args); + } + ); + sink = createMemorySink(); + }); + + afterEach(() => jest.restoreAllMocks()); + + function makeRequester( + fetch, + params = {}, + RequesterClass = TestRequester + ) { + return new RequesterClass({ + fetch, + requestTimeoutMs: 0, + random: () => 0, + ...params, + }); + } + + const withPolicy = (rateLimit) => { + class PolicyRequester extends TestRequester { + static rateLimit = rateLimit; + } + return PolicyRequester; + }; + + describe('a module that declares nothing', () => { + it('retries a bare 429 on the fixed ladder, then throws a plain FetchError', async () => { + const fetch = throttledFetch(Array(7).fill(limited())); + const requester = makeRequester(fetch); + + const error = await requester._get({ url }).catch((e) => e); + + expect(error).toBeInstanceOf(FetchError); + expect(error).not.toBeInstanceOf(RateLimitError); + expect(error.isRateLimited).toBeUndefined(); + expect(error.statusCode).toBe(429); + expect(delays).toEqual(LADDER_MS); + expect(fetch).toHaveBeenCalledTimes(7); + }); + + it('keeps the ladder inside a short invocation, because only hinted waits are budgeted', async () => { + const fetch = throttledFetch(Array(7).fill(limited())); + const requester = makeRequester(fetch, { + requestTimeoutMs: 45_000, + }); + + const error = await runWithInvocationDeadline( + { deadlineAt: Date.now() + 29_000 }, + () => requester._get({ url }).catch((e) => e) + ); + + expect(error).toBeInstanceOf(FetchError); + expect(delays.filter((delay) => delay !== 45_000)).toEqual( + LADDER_MS + ); + expect(fetch).toHaveBeenCalledTimes(7); + }); + + it('leaves the 5xx ladder alone', async () => { + const fetch = throttledFetch([ + { status: 500 }, + { status: 500 }, + ok, + ]); + const result = await makeRequester(fetch)._get({ url }); + + expect(result).toEqual({ ok: true }); + expect(delays).toEqual([1_000, 3_000]); + }); + + it('keeps the ladder for a 503 that carries Retry-After', async () => { + const fetch = throttledFetch([ + { status: 503, headers: { 'Retry-After': '30' } }, + ok, + ]); + await makeRequester(fetch)._get({ url }); + + expect(delays).toEqual([1_000]); + }); + + it('writes no rate-limit record for the ladder', async () => { + const fetch = throttledFetch([limited(), ok]); + await makeRequester(fetch)._get({ url }); + + expect( + sink.records.filter((r) => + r.eventName?.endsWith('.rate_limited') + ) + ).toEqual([]); + }); + }); + + describe('Retry-After', () => { + it('waits as long as the header says, not one ladder step', async () => { + const fetch = throttledFetch([ + limited({ 'Retry-After': '5' }), + ok, + ]); + const result = await makeRequester(fetch)._get({ url }); + + expect(result).toEqual({ ok: true }); + expect(delays).toEqual([5_000]); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it('does not raise a short Retry-After to the ladder step of a later attempt', async () => { + const fetch = throttledFetch([ + limited(), + limited(), + limited({ 'Retry-After': '2' }), + ok, + ]); + await makeRequester(fetch)._get({ url }); + + expect(delays).toEqual([1_000, 3_000, 2_000]); + }); + + it('reads the header from a real Headers object', async () => { + const fetch = throttledFetch([ + limited(undefined, { + headersObject: new Headers({ 'retry-after': '3' }), + }), + ok, + ]); + await makeRequester(fetch)._get({ url }); + + expect(delays).toEqual([3_000]); + }); + + it('adds jitter that never passes the wait by more than a tenth', async () => { + const fetch = throttledFetch([ + limited({ 'Retry-After': '60' }), + ok, + ]); + await makeRequester(fetch, { random: () => 0.999999 })._get({ + url, + }); + + expect(delays).toEqual([60_999]); + }); + + it('sleeps on for hinted 429s beyond the length of the ladder', async () => { + const fetch = throttledFetch([ + ...Array(8).fill(limited({ 'Retry-After': '1' })), + ok, + ]); + const result = await makeRequester(fetch)._get({ url }); + + expect(result).toEqual({ ok: true }); + expect(delays).toEqual(Array(8).fill(1_000)); + expect(fetch).toHaveBeenCalledTimes(9); + }); + + it('stops after 10 hinted retries even when every wait fits the cap', async () => { + const fetch = throttledFetch( + Array(11).fill(limited({ 'Retry-After': '1' })) + ); + + const error = await makeRequester(fetch, { backOff: [1, 2] }) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect(error.statusCode).toBe(429); + expect(delays).toEqual(Array(10).fill(1_000)); + expect(fetch).toHaveBeenCalledTimes(11); + }); + + it('keeps the whole 5xx ladder after a hinted retry', async () => { + const fetch = throttledFetch([ + limited({ 'Retry-After': '1' }), + ...Array(7).fill({ status: 500 }), + ]); + + const error = await makeRequester(fetch) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(FetchError); + expect(error.statusCode).toBe(500); + expect(delays).toEqual([1_000, ...LADDER_MS]); + expect(fetch).toHaveBeenCalledTimes(8); + }); + + it('does not count a 5xx retry or a connection reset as a hinted retry', async () => { + const reset = Object.assign(new Error('socket hang up'), { + code: 'ECONNRESET', + }); + const fetch = throttledFetch([ + ...Array(9).fill(limited({ 'Retry-After': '1' })), + { status: 500 }, + { reject: reset }, + limited({ 'Retry-After': '1' }), + ok, + ]); + + const result = await makeRequester(fetch)._get({ url }); + + expect(result).toEqual({ ok: true }); + expect(fetch).toHaveBeenCalledTimes(13); + }); + + it('throws a RateLimitError with retryAt when the wait is over the cap', async () => { + const fetch = throttledFetch([ + limited({ 'Retry-After': '400' }), + ok, + ]); + const before = Date.now(); + + const error = await makeRequester(fetch) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect(error).toBeInstanceOf(FetchError); + expect(error).toMatchObject({ + isRateLimited: true, + statusCode: 429, + source: 'header', + waitMs: 400_000, + reason: 'unknown', + }); + expect(error.retryAt.getTime()).toBeGreaterThanOrEqual( + before + 400_000 + ); + expect(fetch).toHaveBeenCalledTimes(1); + expect(delays).toEqual([]); + }); + + it('stops sleeping once the waits add up to the cap', async () => { + const fetch = throttledFetch([ + ...Array(4).fill(limited({ 'Retry-After': '100' })), + ok, + ]); + + const error = await makeRequester(fetch) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect(delays).toEqual([100_000, 100_000, 100_000]); + expect(fetch).toHaveBeenCalledTimes(4); + }); + + it('counts a hinted wait against the cap across a 5xx retry', async () => { + const fetch = throttledFetch([ + limited({ 'Retry-After': '200' }), + { status: 500 }, + limited({ 'Retry-After': '200' }), + ok, + ]); + + const error = await makeRequester(fetch) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect(delays).toEqual([200_000, 1_000]); + expect(fetch).toHaveBeenCalledTimes(3); + }); + + it('counts a hinted wait against the cap across a connection reset', async () => { + const reset = Object.assign(new Error('socket hang up'), { + code: 'ECONNRESET', + }); + const fetch = throttledFetch([ + limited({ 'Retry-After': '200' }), + { reject: reset }, + limited({ 'Retry-After': '200' }), + ok, + ]); + + const error = await makeRequester(fetch) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect(delays).toEqual([200_000, 1_000]); + }); + + it('sleeps a short wait inside an invocation shorter than the default request timeout', async () => { + const fetch = throttledFetch([ + limited({ 'Retry-After': '2' }), + ok, + ]); + const requester = makeRequester(fetch, { + requestTimeoutMs: undefined, + }); + + const result = await runWithInvocationDeadline( + { deadlineAt: Date.now() + 29_000 }, + () => requester._get({ url }) + ); + + expect(requester.requestTimeoutMs).toBe(60_000); + expect(result).toEqual({ ok: true }); + expect(delays.filter((delay) => delay !== 60_000)).toEqual([ + 2_000, + ]); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it('throws at once when the wait does not fit the time left in the invocation', async () => { + const fetch = throttledFetch([ + limited({ 'Retry-After': '5' }), + ok, + ]); + const requester = makeRequester(fetch, { + requestTimeoutMs: 5_000, + }); + + const error = await runWithInvocationDeadline( + { deadlineAt: Date.now() + 8_000 }, + () => requester._get({ url }).catch((e) => e) + ); + + expect(error).toBeInstanceOf(RateLimitError); + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it('throws at once when the caller allows no in-process wait', async () => { + const fetch = throttledFetch([ + limited({ 'Retry-After': '2' }), + ok, + ]); + + const error = await runWithInvocationDeadline( + { maxInProcessWaitMs: 0 }, + () => + makeRequester(fetch) + ._get({ url }) + .catch((e) => e) + ); + + expect(error).toBeInstanceOf(RateLimitError); + expect(fetch).toHaveBeenCalledTimes(1); + expect(delays).toEqual([]); + }); + + it('sleeps only as long as the budget of the caller allows', async () => { + const fetch = throttledFetch([ + ...Array(3).fill(limited({ 'Retry-After': '10' })), + ok, + ]); + + const error = await runWithInvocationDeadline( + { maxInProcessWaitMs: 25_000 }, + () => + makeRequester(fetch) + ._get({ url }) + .catch((e) => e) + ); + + expect(error).toBeInstanceOf(RateLimitError); + expect(delays).toEqual([10_000, 10_000]); + expect(fetch).toHaveBeenCalledTimes(3); + }); + + it('ignores a maxInProcessWaitMs in the static rateLimit of the module', async () => { + const Api = withPolicy({ maxInProcessWaitMs: 0 }); + const fetch = throttledFetch([ + limited({ 'Retry-After': '2' }), + ok, + ]); + + const result = await makeRequester(fetch, {}, Api)._get({ + url, + }); + + expect(result).toEqual({ ok: true }); + expect(delays).toEqual([2_000]); + }); + + it('reads the body of the RateLimitError once', async () => { + const fetch = throttledFetch([ + limited( + { 'Retry-After': '400' }, + { body: { message: 'slow down' } } + ), + ]); + const error = await makeRequester(fetch) + ._get({ url }) + .catch((e) => e); + + expect(fetch.responses[0].text).toHaveBeenCalledTimes(1); + expect(error.body).toContain('slow down'); + }); + }); + + describe('with an empty backOff', () => { + it('throws a RateLimitError at once for a Retry-After', async () => { + const fetch = throttledFetch([ + limited({ 'Retry-After': '2' }), + ok, + ]); + + const error = await makeRequester(fetch, { backOff: [] }) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect(error).toMatchObject({ + statusCode: 429, + source: 'header', + }); + expect(fetch).toHaveBeenCalledTimes(1); + expect(delays).toEqual([]); + }); + + it('throws a RateLimitError at once for a hint that classify gives', async () => { + const Api = withPolicy({ + classify: () => ({ reason: 'daily', waitMs: 2_000 }), + }); + const fetch = throttledFetch([ + { status: 403, body: { code: 'LIMIT' } }, + ok, + ]); + + const error = await makeRequester(fetch, { backOff: [] }, Api) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect(error).toMatchObject({ + statusCode: 403, + reason: 'daily', + }); + expect(fetch).toHaveBeenCalledTimes(1); + expect(delays).toEqual([]); + }); + + it('still throws a plain FetchError for a 429 without a hint', async () => { + const fetch = throttledFetch([limited(), ok]); + + const error = await makeRequester(fetch, { backOff: [] }) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(FetchError); + expect(error).not.toBeInstanceOf(RateLimitError); + expect(error.statusCode).toBe(429); + expect(fetch).toHaveBeenCalledTimes(1); + }); + }); + + describe('static rateLimit', () => { + it('makes minRetryAfterMs the wait of a bare 429', async () => { + const Api = withPolicy({ minRetryAfterMs: 60_000 }); + const fetch = throttledFetch([limited(), ok]); + await makeRequester(fetch, {}, Api)._get({ url }); + + expect(delays).toEqual([60_000]); + }); + + it('makes minRetryAfterMs the floor of a shorter Retry-After', async () => { + const Api = withPolicy({ minRetryAfterMs: 60_000 }); + const fetch = throttledFetch([ + limited({ 'Retry-After': '5' }), + ok, + ]); + await makeRequester(fetch, {}, Api)._get({ url }); + + expect(delays).toEqual([60_000]); + }); + + it('reads only the parsers it lists, in that order', async () => { + const Api = withPolicy({ + parsers: ['resetHeaders', 'retryAfter'], + }); + const fetch = throttledFetch([ + limited({ 'Retry-After': '2', 'X-RateLimit-Reset': '7' }), + ok, + ]); + await makeRequester(fetch, {}, Api)._get({ url }); + + expect(delays).toEqual([7_000]); + }); + + it('carries the module name and the scope key on the error', async () => { + const Api = withPolicy({ scope: 'entity' }); + const fetch = throttledFetch([ + limited({ 'Retry-After': '400' }), + ]); + const requester = makeRequester( + fetch, + { delegate: { name: 'hubspot', entity: { id: 7 } } }, + Api + ); + + const error = await requester._get({ url }).catch((e) => e); + + expect(error).toMatchObject({ + module: 'hubspot', + scopeKey: 'hubspot:entity:7', + }); + }); + + it('fails at construction for a parser that does not exist', () => { + const Api = withPolicy({ parsers: ['magic'] }); + expect(() => new Api({})).toThrow(TypeError); + }); + }); + + describe('classify', () => { + it('treats a 403 as throttled and retries when the wait fits', async () => { + const classify = jest.fn(({ status, body }) => + status === 403 && body?.code === 'LIMIT' + ? { reason: 'daily', waitMs: 2_000 } + : null + ); + const Api = withPolicy({ classify }); + const fetch = throttledFetch([ + { status: 403, body: { code: 'LIMIT' } }, + ok, + ]); + + const result = await makeRequester(fetch, {}, Api)._get({ + url, + }); + + expect(result).toEqual({ ok: true }); + expect(delays).toEqual([2_000]); + expect(classify).toHaveBeenCalledWith( + expect.objectContaining({ + status: 403, + body: { code: 'LIMIT' }, + }) + ); + }); + + it('throws a RateLimitError with the 403 status when the wait does not fit', async () => { + const Api = withPolicy({ + classify: () => ({ reason: 'daily', waitMs: 400_000 }), + }); + const fetch = throttledFetch([ + { status: 403, body: { code: 'LIMIT' } }, + ]); + + const error = await makeRequester(fetch, {}, Api) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect(error).toMatchObject({ + statusCode: 403, + reason: 'daily', + source: 'body', + }); + expect(fetch.responses[0].text).toHaveBeenCalledTimes(1); + expect(error.body).toContain('LIMIT'); + }); + + it('leaves a 403 that classify does not recognise as a plain FetchError', async () => { + const Api = withPolicy({ classify: () => null }); + const fetch = throttledFetch([ + { status: 403, headers: { 'Retry-After': '2' } }, + ]); + + const error = await makeRequester(fetch, {}, Api) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(FetchError); + expect(error.isRateLimited).toBeUndefined(); + expect(fetch).toHaveBeenCalledTimes(1); + expect(delays).toEqual([]); + }); + + it('throws a RateLimitError at once for a 403 that classify names by reason only', async () => { + const Api = withPolicy({ + classify: () => ({ reason: 'daily' }), + }); + const fetch = throttledFetch( + Array(7).fill({ status: 403, body: { code: 'LIMIT' } }) + ); + + const error = await makeRequester(fetch, {}, Api) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect(error).toMatchObject({ + statusCode: 403, + isRateLimited: true, + reason: 'daily', + }); + expect(error.retryAt).toBeUndefined(); + expect(error.waitMs).toBeUndefined(); + expect(error.module).toEqual(expect.any(String)); + expect(fetch).toHaveBeenCalledTimes(1); + expect(delays).toEqual([]); + }); + + it.each([ + 'application/problem+json', + 'Application/JSON; charset=utf-8', + 'application/vnd.api+json', + 'application/hal+json', + 'text/json', + ])( + 'hands classify the parsed body of a %s response', + async (contentType) => { + const classify = jest.fn(() => ({ + reason: 'burst', + waitMs: 1_000, + })); + const Api = withPolicy({ classify }); + const fetch = throttledFetch([ + { status: 403, contentType, body: { code: 'LIMIT' } }, + ok, + ]); + + await makeRequester(fetch, {}, Api)._get({ url }); + + expect(classify.mock.calls[0][0].body).toEqual({ + code: 'LIMIT', + }); + } + ); + + it.each(['text/plain', 'application/x-amz-json-1.1'])( + 'hands classify the parsed body of JSON sent as %s', + async (contentType) => { + const classify = jest.fn(() => ({ + reason: 'burst', + waitMs: 1_000, + })); + const Api = withPolicy({ classify }); + const fetch = throttledFetch([ + { status: 403, contentType, text: '{"code":"LIMIT"}' }, + ok, + ]); + + await makeRequester(fetch, {}, Api)._get({ url }); + + expect(classify.mock.calls[0][0].body).toEqual({ + code: 'LIMIT', + }); + } + ); + + it('hands classify an undefined body for an HTML error page', async () => { + const classify = jest.fn(() => null); + const Api = withPolicy({ classify }); + const fetch = throttledFetch([ + { + status: 403, + contentType: 'text/html', + text: 'Forbidden', + }, + ]); + + const error = await makeRequester(fetch, {}, Api) + ._get({ url }) + .catch((e) => e); + + expect(classify.mock.calls[0][0].body).toBeUndefined(); + expect(error.body).toBe('Forbidden'); + }); + + it('sleeps the Retry-After of a reason-only classify and keeps its reason', async () => { + const Api = withPolicy({ + classify: () => ({ reason: 'burst' }), + }); + const fetch = throttledFetch([ + limited({ 'Retry-After': '7' }), + ok, + ]); + + const result = await makeRequester(fetch, {}, Api)._get({ + url, + }); + + expect(result).toEqual({ ok: true }); + expect(delays).toEqual([7_000]); + expect( + sink.records.filter( + (r) => + r.eventName === + 'module.PolicyRequester.rate_limited' + ) + ).toEqual([ + expect.objectContaining({ + waitMs: 7_000, + reason: 'burst', + hintSource: 'header', + }), + ]); + }); + + it('takes the time of a reason-only classify from Retry-After', async () => { + const Api = withPolicy({ + classify: () => ({ reason: 'burst' }), + }); + const fetch = throttledFetch([ + limited({ 'Retry-After': '400' }), + ]); + + const error = await makeRequester(fetch, {}, Api) + ._get({ url }) + .catch((e) => e); + + expect(error).toMatchObject({ + reason: 'burst', + source: 'header', + }); + }); + + it('hands classify an undefined body when the response is not JSON', async () => { + const classify = jest.fn(() => ({ + reason: 'burst', + waitMs: 1_000, + })); + const Api = withPolicy({ classify }); + const fetch = throttledFetch([ + limited(undefined, { + contentType: 'text/plain', + text: 'slow down', + }), + ok, + ]); + + await makeRequester(fetch, {}, Api)._get({ url }); + + expect(classify.mock.calls[0][0].body).toBeUndefined(); + }); + + it('does not retry or flag a 404 when classify returns a Promise', async () => { + const Api = withPolicy({ + classify: async () => ({ reason: 'daily' }), + }); + const fetch = throttledFetch([{ status: 404 }, ok]); + + const error = await makeRequester(fetch, {}, Api) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(FetchError); + expect(error.statusCode).toBe(404); + expect(error.isRateLimited).toBeUndefined(); + expect(fetch).toHaveBeenCalledTimes(1); + expect(delays).toEqual([]); + expect( + sink.records.filter( + (r) => + r.eventName === + 'module.PolicyRequester.rate_limit_classify_failed' + ) + ).toEqual([ + expect.objectContaining({ + level: 'WARN', + statusCode: 404, + error: expect.objectContaining({ + message: + 'classify() must return a hint or null, not a Promise', + }), + }), + ]); + }); + + it('does not call classify for a success or for a 401', async () => { + const classify = jest.fn(() => null); + const Api = withPolicy({ classify }); + const fetch = throttledFetch([{ status: 401 }, ok]); + const requester = makeRequester(fetch, { backOff: [0] }, Api); + requester.notify = jest.fn(); + + await requester._get({ url }); + + expect(classify).not.toHaveBeenCalled(); + }); + + it('writes one WARN record and goes on with the parsers when classify throws', async () => { + const Api = withPolicy({ + classify: () => { + throw new Error('bad classifier'); + }, + }); + const fetch = throttledFetch([ + limited({ 'Retry-After': '2' }), + ok, + ]); + + await makeRequester(fetch, {}, Api)._get({ url }); + + expect(delays).toEqual([2_000]); + expect( + sink.records.filter( + (r) => + r.eventName === + 'module.PolicyRequester.rate_limit_classify_failed' + ) + ).toEqual([ + expect.objectContaining({ + level: 'WARN', + statusCode: 429, + error: expect.objectContaining({ + message: 'bad classifier', + }), + }), + ]); + }); + }); + + describe('logging', () => { + it('writes one TRACE record for a wait that a header set', async () => { + const fetch = throttledFetch([ + limited({ 'Retry-After': '5' }), + ok, + ]); + await makeRequester(fetch)._get({ url }); + + const records = sink.records.filter( + (r) => r.eventName === 'module.TestRequester.rate_limited' + ); + expect(records).toEqual([ + expect.objectContaining({ + level: 'TRACE', + logger: 'module.TestRequester', + statusCode: 429, + waitMs: 5_000, + reason: 'unknown', + hintSource: 'header', + attempt: 0, + waitedMs: 0, + retryAt: expect.any(String), + }), + ]); + expect(records[0]).not.toHaveProperty('action'); + }); + + it('writes no rate-limit record for the error it throws', async () => { + const fetch = throttledFetch([ + limited({ 'Retry-After': '400' }), + ]); + const error = await makeRequester(fetch) + ._get({ url }) + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect( + sink.records.filter((r) => + r.eventName?.endsWith('.rate_limited') + ) + ).toEqual([]); + }); }); }); @@ -238,16 +1117,11 @@ describe('Requester', () => { headers: { get: () => 'application/json' }, json: () => new Promise((_resolve, reject) => { - capturedSignal.addEventListener( - 'abort', - () => { - const err = new Error( - 'aborted mid-body' - ); - err.name = 'AbortError'; - reject(err); - } - ); + capturedSignal.addEventListener('abort', () => { + const err = new Error('aborted mid-body'); + err.name = 'AbortError'; + reject(err); + }); }), }; }); @@ -441,11 +1315,15 @@ describe('Requester', () => { requester.refreshAuth = jest.fn().mockResolvedValue(true); requester.notify = jest.fn(); - const result = await requester._get({ url: 'https://example.com/protected' }); + const result = await requester._get({ + url: 'https://example.com/protected', + }); expect(result).toEqual({ ok: true }); expect(requester.refreshAuth).toHaveBeenCalledTimes(1); - expect(requester.notify).not.toHaveBeenCalledWith(requester.DLGT_INVALID_AUTH); + expect(requester.notify).not.toHaveBeenCalledWith( + requester.DLGT_INVALID_AUTH + ); expect(fetchMock).toHaveBeenCalledTimes(2); }); @@ -524,13 +1402,19 @@ describe('Requester', () => { requester.refreshAuth = jest.fn().mockResolvedValue(true); requester.notify = jest.fn(); - const r1 = await requester._get({ url: 'https://example.com/first' }); - const r2 = await requester._get({ url: 'https://example.com/second' }); + const r1 = await requester._get({ + url: 'https://example.com/first', + }); + const r2 = await requester._get({ + url: 'https://example.com/second', + }); expect(r1).toEqual({ ok: 'first' }); expect(r2).toEqual({ ok: 'second' }); expect(requester.refreshAuth).toHaveBeenCalledTimes(2); - expect(requester.notify).not.toHaveBeenCalledWith(requester.DLGT_INVALID_AUTH); + expect(requester.notify).not.toHaveBeenCalledWith( + requester.DLGT_INVALID_AUTH + ); expect(fetchMock).toHaveBeenCalledTimes(4); }); }); diff --git a/packages/core/types/core/index.d.ts b/packages/core/types/core/index.d.ts index 2ac09c449..084ac9723 100644 --- a/packages/core/types/core/index.d.ts +++ b/packages/core/types/core/index.d.ts @@ -1,5 +1,10 @@ declare module "@friggframework/core" { import type { SendMessageCommandInput } from "@aws-sdk/client-sqs"; + import type { RateLimitHint } from "@friggframework/errors"; + import type { + RateLimitPolicy, + RateLimitSignal, + } from "@friggframework/module-plugin"; export class Delegate implements IFriggDelegate { delegate: any; @@ -53,6 +58,34 @@ declare module "@friggframework/core" { export function loadInstalledModules(): any[]; + export function runWithInvocationDeadline( + limits: { deadlineAt?: number; maxInProcessWaitMs?: number }, + fn: () => T + ): T; + + export function remainingInvocationMs(now?: number): number; + + export function classifyRateLimit( + policy: RateLimitPolicy | undefined, + signal: RateLimitSignal, + options?: { now?: number; onClassifyError?: (error: Error) => void } + ): RateLimitHint | null; + + export function parseRetryAfter( + value: string | null | undefined, + options?: { now?: number } + ): RateLimitHint | null; + + export function parseResetHeaders( + headers: object | undefined, + options?: { now?: number } + ): RateLimitHint | null; + + export function parseIetfRateLimit( + headers: object | undefined, + options?: { now?: number } + ): RateLimitHint | null; + type GetQueueURLParams = { QueueName: string; QueueOwnerAWSAccountId?: string; diff --git a/packages/core/types/errors/index.d.ts b/packages/core/types/errors/index.d.ts index 5a321723d..f0cc21dad 100644 --- a/packages/core/types/errors/index.d.ts +++ b/packages/core/types/errors/index.d.ts @@ -16,10 +16,57 @@ declare module "@friggframework/errors" { readonly body: any; isTimeout?: boolean; timeoutMs?: number; + isRateLimited?: boolean; + reason?: RateLimitReason; static create(options?: CreateFetchErrorParams): Promise; } + export type RateLimitReason = + | "burst" + | "daily" + | "monthly" + | "concurrency" + | "unknown"; + + export type RateLimitSource = "header" | "body" | "static" | "backoff"; + + export type RateLimitHint = { + retryAt: Date; + waitMs: number; + reason: RateLimitReason; + policy?: string; + remaining?: number; + source: RateLimitSource; + }; + + export class RateLimitError extends FetchError { + constructor(options?: RateLimitErrorConstructor); + + isRateLimited: true; + retryAt?: Date; + waitMs?: number; + reason: RateLimitReason; + policy?: string; + source: RateLimitSource | "unknown"; + module?: string; + scopeKey?: string; + + static create( + options?: CreateRateLimitErrorParams + ): Promise; + } + + type RateLimitErrorConstructor = FetchErrorConstructor & { + hint?: Partial; + waitMs?: number; + module?: string; + scopeKey?: string; + now?: number; + }; + + type CreateRateLimitErrorParams = RateLimitErrorConstructor; + type FetchErrorResponse = { headers?: object; status?: number; diff --git a/packages/core/types/module-plugin/index.d.ts b/packages/core/types/module-plugin/index.d.ts index 8eaf41f75..0c468c552 100644 --- a/packages/core/types/module-plugin/index.d.ts +++ b/packages/core/types/module-plugin/index.d.ts @@ -1,5 +1,6 @@ declare module "@friggframework/module-plugin" { import { Delegate, IFriggDelegate } from "@friggframework/core"; + import type { RateLimitHint } from "@friggframework/errors"; export interface Credential { id?: string; @@ -22,8 +23,49 @@ declare module "@friggframework/module-plugin" { export type MappedEntity = Entity & { id: string; type: any }; + export type RateLimitScope = + | "credential" + | "entity" + | "app" + | ((requester: Requester) => string | number | null | undefined); + + export type RateLimitSignal = { + status?: number; + headers?: { get(name: string): string | null } | object; + body?: unknown; + }; + + export type RateLimitWindow = { + name: string; + limit?: number; + perMs?: number; + rollingMs?: number; + resets?: { at: string; tz: string }; + }; + + export type RateLimitPolicy = { + scope?: RateLimitScope; + windows?: RateLimitWindow[]; + maxConcurrency?: number; + minRetryAfterMs?: number; + parsers?: Array<"retryAfter" | "resetHeaders" | "ietf">; + classify?( + signal: RateLimitSignal + ): Partial> & { + source?: "header" | "body" | "static"; + } | null | undefined; + userHints?: Record< + string, + { links?: Array<{ label: string; url: string }> } + >; + }; + export class Requester implements IFriggRequester { + static rateLimit?: RateLimitPolicy; + static requestTimeoutMs?: number; + DLGT_INVALID_AUTH: string; + requestTimeoutMs: number; backOff: number[]; fetch: any; isRefreshable: boolean; @@ -87,6 +129,7 @@ declare module "@friggframework/module-plugin" { type RequesterConstructor = { backOff?: number[]; fetch?: any; + random?: () => number; }; export class ApiKeyRequester