From 5e856d32fde22f7a94b088d8be588a00dfbd1cc5 Mon Sep 17 00:00:00 2001 From: d-klotz Date: Tue, 29 Sep 2026 19:10:36 -0300 Subject: [PATCH 1/3] feat(hubspot): declare the rate-limit policy HubSpot limits each account that installs a marketplace OAuth app to 110 requests per 10 seconds, and answers a throttled call with a 429 whose body names the limit in `policyName`. The Requester met every such 429 with its fixed 1, 3, 10, 30, 60 and 180 second ladder, whatever the response said. Declare `static rateLimit` so the Requester can use what the response says: - A burst 429 (`TEN_SECONDLY_ROLLING`) returns a reason-only hint. A real `Retry-After` then wins, and the 10 second burst window is the fallback. - A `DAILY` 429 returns a one hour probe (`source: 'static'`). HubSpot resets the daily limit at midnight in the account's time zone, which the module does not know, so no UTC reset time is computed. One hour is over the in-process budget, so the Requester throws a `RateLimitError` with `reason: 'daily'` and `retryAt` and does not sleep. - Any other status is not a limit. - `userHints.daily.links` points at the API usage guidelines. `maxConcurrency` (10) is a pacing figure that HubSpot does not document. It is declared for the pacing phase and nothing reads it yet. The tests use recorded-style 429 responses and a stubbed fetch. They need a `@friggframework/core` that exports `RateLimitError` (friggframework/frigg#656). The dependency range is unchanged until the release that ships it is known. Refs ADR-049. Co-Authored-By: Claude Sonnet 5.5 Co-Authored-By: Claude Opus 5.5 --- packages/v1-ready/hubspot/README.md | 20 ++ packages/v1-ready/hubspot/api.js | 31 +++ .../v1-ready/hubspot/tests/rateLimit.test.js | 231 ++++++++++++++++++ 3 files changed, 282 insertions(+) create mode 100644 packages/v1-ready/hubspot/tests/rateLimit.test.js diff --git a/packages/v1-ready/hubspot/README.md b/packages/v1-ready/hubspot/README.md index c2ad5b2..3bec4dc 100644 --- a/packages/v1-ready/hubspot/README.md +++ b/packages/v1-ready/hubspot/README.md @@ -113,3 +113,23 @@ colliding. - **`HUBSPOT_WEBHOOK`** — queue event (worker, hydrated integration): default no-op; override via `binding.handlers.HUBSPOT_WEBHOOK` to run your per-event logic with the correct per-account context. + +## Rate limits + +`Api.rateLimit` tells the Frigg Requester how HubSpot limits calls. The numbers come +from the [API usage guidelines and limits](https://developers.hubspot.com/docs/developer-tooling/platform/usage-guidelines). +A `@friggframework/core` release without rate-limit awareness ignores the declaration. + +- **Burst:** each account that installs a marketplace OAuth app is limited to 110 + requests every 10 seconds. A throttled call returns `429` with `policyName` + `TEN_SECONDLY_ROLLING`. The Requester waits for `Retry-After` when the response has + one, and for the 10 second window when it has none. +- **Daily:** a `429` with `policyName` `DAILY`. HubSpot resets the daily limit at + midnight in the account's time zone, which the module does not know, so it does not + compute a reset time. It reports a one hour probe interval: the Requester throws a + `RateLimitError` with `reason: 'daily'` and a `retryAt` one hour ahead. It does not + sleep for an hour in process. +- **Any other status** is not a limit and keeps its normal handling. + +HubSpot documents no `Retry-After` or reset header for these responses, and the search +endpoints send no `X-HubSpot-RateLimit-*` headers at all. diff --git a/packages/v1-ready/hubspot/api.js b/packages/v1-ready/hubspot/api.js index 9276811..9eafa7f 100644 --- a/packages/v1-ready/hubspot/api.js +++ b/packages/v1-ready/hubspot/api.js @@ -1,6 +1,37 @@ const {OAuth2Requester, get} = require('@friggframework/core'); +const DAILY_PROBE_INTERVAL_MS = 3_600_000; + class Api extends OAuth2Requester { + static rateLimit = { + scope: 'entity', + windows: [{ name: 'burst', limit: 110, perMs: 10_000 }], + maxConcurrency: 10, + parsers: ['retryAfter', 'resetHeaders'], + classify({ status, body }) { + if (status !== 429) return null; + if (body?.policyName === 'DAILY') { + return { + reason: 'daily', + policy: 'DAILY', + waitMs: DAILY_PROBE_INTERVAL_MS, + source: 'static', + }; + } + return { reason: 'burst', policy: body?.policyName }; + }, + userHints: { + daily: { + links: [ + { + label: 'API usage guidelines and limits', + url: 'https://developers.hubspot.com/docs/developer-tooling/platform/usage-guidelines', + }, + ], + }, + }, + }; + constructor(params) { super(params); // The majority of the properties for OAuth are default loaded by OAuth2Requester. diff --git a/packages/v1-ready/hubspot/tests/rateLimit.test.js b/packages/v1-ready/hubspot/tests/rateLimit.test.js new file mode 100644 index 0000000..4c24d36 --- /dev/null +++ b/packages/v1-ready/hubspot/tests/rateLimit.test.js @@ -0,0 +1,231 @@ +const { FetchError, RateLimitError } = require('@friggframework/core'); +const { Api } = require('../api'); + +const ONE_HOUR_MS = 3_600_000; + +const burstLimitBody = { + status: 'error', + message: 'You have reached your ten_secondly_rolling limit.', + errorType: 'RATE_LIMIT', + correlationId: '00000000-0000-4000-8000-000000000000', + policyName: 'TEN_SECONDLY_ROLLING', +}; + +const dailyLimitBody = { + status: 'error', + message: 'You have reached your daily limit.', + errorType: 'RATE_LIMIT', + correlationId: '00000000-0000-4000-8000-000000000000', + policyName: 'DAILY', +}; + +const burstHeaders = { + 'X-HubSpot-RateLimit-Max': '110', + 'X-HubSpot-RateLimit-Remaining': '0', + 'X-HubSpot-RateLimit-Interval-Milliseconds': '10000', +}; + +function throttledFetch(responses) { + const queue = [...responses]; + return jest.fn(async () => { + const next = queue.shift(); + if (!next) throw new Error('unexpected fetch call'); + const body = next.body ?? {}; + return { + status: next.status, + bodyUsed: false, + headers: new Map( + Object.entries({ + 'Content-Type': next.contentType ?? 'application/json', + ...next.headers, + }) + ), + json: async () => body, + text: jest.fn(async () => next.text ?? JSON.stringify(body)), + }; + }); +} + +const limited = (body, headers) => ({ status: 429, body, headers }); +const ok = { status: 200, body: { results: [] } }; + +describe('HubSpot rate limits', () => { + let delays; + + beforeEach(() => { + delays = []; + const realSetTimeout = global.setTimeout; + jest.spyOn(global, 'setTimeout').mockImplementation( + (fn, delay, ...args) => { + delays.push(delay); + return realSetTimeout(fn, 0, ...args); + } + ); + jest.spyOn(Math, 'random').mockReturnValue(0); + }); + + afterEach(() => jest.restoreAllMocks()); + + const makeApi = (fetch, ApiClass = Api) => + new ApiClass({ + fetch, + requestTimeoutMs: 0, + delegate: { name: 'hubspot', entity: { id: 7 } }, + }); + + describe('a burst 429', () => { + it('waits as long as Retry-After says, then retries', async () => { + const fetch = throttledFetch([ + limited(burstLimitBody, { + ...burstHeaders, + 'Retry-After': '2', + }), + ok, + ]); + + const result = await makeApi(fetch).listCompanies(); + + expect(result).toEqual({ results: [] }); + expect(delays).toEqual([2_000]); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it('waits the 10 second burst window when the response has no Retry-After', async () => { + const fetch = throttledFetch([ + limited(burstLimitBody, burstHeaders), + ok, + ]); + + const result = await makeApi(fetch).listCompanies(); + + expect(result).toEqual({ results: [] }); + expect(delays).toEqual([10_000]); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it('keeps the burst reason and the policy name when a long Retry-After makes it throw', async () => { + const fetch = throttledFetch([ + limited(burstLimitBody, { 'Retry-After': '400' }), + ok, + ]); + + const error = await makeApi(fetch) + .listCompanies() + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect(error).toMatchObject({ + reason: 'burst', + policy: 'TEN_SECONDLY_ROLLING', + source: 'header', + waitMs: 400_000, + }); + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it('waits the burst window when the body is not JSON', async () => { + const fetch = throttledFetch([ + { + status: 429, + contentType: 'text/html', + text: 'Too Many Requests', + }, + ok, + ]); + + await makeApi(fetch).listCompanies(); + + expect(delays).toEqual([10_000]); + }); + }); + + describe('a DAILY 429', () => { + it('throws a RateLimitError that probes again in one hour', async () => { + const fetch = throttledFetch([limited(dailyLimitBody), ok]); + const before = Date.now(); + + const error = await makeApi(fetch) + .listCompanies() + .catch((e) => e); + + expect(error).toBeInstanceOf(RateLimitError); + expect(error).toBeInstanceOf(FetchError); + expect(error).toMatchObject({ + isRateLimited: true, + statusCode: 429, + reason: 'daily', + policy: 'DAILY', + source: 'static', + waitMs: ONE_HOUR_MS, + module: 'hubspot', + scopeKey: 'hubspot:entity:7', + }); + expect(error.retryAt.getTime()).toBeGreaterThanOrEqual( + before + ONE_HOUR_MS + ); + expect(error.retryAt.getTime()).toBeLessThanOrEqual( + Date.now() + ONE_HOUR_MS + ); + expect(fetch).toHaveBeenCalledTimes(1); + expect(delays).toEqual([]); + }); + + it('sleeps the probe interval and retries when the in-process budget allows it', async () => { + class PatientApi extends Api { + static rateLimit = { + ...Api.rateLimit, + maxInProcessWaitMs: ONE_HOUR_MS + 60_000, + }; + } + const fetch = throttledFetch([limited(dailyLimitBody), ok]); + + const result = await makeApi(fetch, PatientApi).listCompanies(); + + expect(result).toEqual({ results: [] }); + expect(delays).toEqual([ONE_HOUR_MS]); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it('links the daily reason to the HubSpot usage guidelines', () => { + const { links } = Api.rateLimit.userHints.daily; + + expect(links).toEqual([ + { + label: expect.any(String), + url: 'https://developers.hubspot.com/docs/developer-tooling/platform/usage-guidelines', + }, + ]); + }); + }); + + describe('other failures', () => { + it('does not treat a 403 as a limit', async () => { + const fetch = throttledFetch([ + { + status: 403, + body: { status: 'error', category: 'MISSING_SCOPES' }, + }, + ok, + ]); + + const error = await makeApi(fetch) + .listCompanies() + .catch((e) => e); + + expect(error).toBeInstanceOf(FetchError); + expect(error).not.toBeInstanceOf(RateLimitError); + expect(error.statusCode).toBe(403); + expect(fetch).toHaveBeenCalledTimes(1); + expect(delays).toEqual([]); + }); + }); + + it('declares its limits', () => { + expect(Api.rateLimit).toMatchObject({ + scope: 'entity', + windows: [{ name: 'burst', limit: 110, perMs: 10_000 }], + maxConcurrency: 10, + parsers: ['retryAfter', 'resetHeaders'], + }); + }); +}); From 953b606cc679df418d4f2d6c437c219fc075a8e2 Mon Sep 17 00:00:00 2001 From: d-klotz Date: Tue, 29 Sep 2026 19:10:36 -0300 Subject: [PATCH 2/3] feat(salesforce): classify the API request and concurrency limits Salesforce counts the API calls of an org over a rolling 24 hours. An org over its limit gets a 403 whose error code is `REQUEST_LIMIT_EXCEEDED`, and no reset time. The module drives jsforce, not the Requester, and a jsforce error has no `statusCode` (`HttpApiError`: `name` equals `errorCode`), so the Requester never sees it and the queue worker gets an opaque error. `REQUEST_LIMIT_EXCEEDED` is also the error for too many concurrent requests that run 20 seconds or longer, which clears in seconds. Reporting it as a daily limit would put a sync on hold for an hour. - Declare `static rateLimit`: a per-org daily rolling window and a `classify` that recognises `REQUEST_LIMIT_EXCEEDED`. Salesforce documents no field that tells the two limits apart, so `classify` reads the message: one that contains "concurrent" is a concurrency limit with a 30 second wait, and any other message is a daily limit. The message text is not documented, so the default stays daily. With no reset time to read, a daily limit gets a static one hour probe. Both reasons link to the API request limits page. - Add the public `withLimits(call)`. On a failed call it runs `classifyRateLimit` and throws a `RateLimitError`, with the jsforce error kept as `cause`. It rethrows any other error unchanged. The data methods use it. A consumer that calls `api.conn` directly wraps its own calls with `api.withLimits(() => api.conn.query(soql))`, as the README says. - Before it throws, `withLimits` tells the delegate with `_notifyRateLimited`, as the Requester does when it throws a `RateLimitError`, so the integration can record a warning for the user. Core writes that warning only for a limit longer than its in-process cap, so the daily limit gets one and the 30 second concurrency limit does not. A core without that method skips the call. - `module` on the error is `_telemetryModuleLabel()`, the label the Requester puts on its own `RateLimitError`. An API instance has no `name`. `RateLimitError` and `classifyRateLimit` come from `@friggframework/core` (friggframework/frigg#656) and are read at runtime. The dependency range is unchanged. On a core that does not export both yet, `withLimits` rethrows the original jsforce error, so the module can be released before an app upgrades its core. The token and refresh code is untouched, and `conn.limitInfo` is not read yet. Refs ADR-049. Co-Authored-By: Claude Sonnet 5.5 Co-Authored-By: Claude Opus 5.5 --- packages/v1-ready/salesforce/README.md | 18 +- packages/v1-ready/salesforce/api.js | 98 +++++- packages/v1-ready/salesforce/test/api.test.js | 294 ++++++++++++++++++ 3 files changed, 398 insertions(+), 12 deletions(-) diff --git a/packages/v1-ready/salesforce/README.md b/packages/v1-ready/salesforce/README.md index 0cc28d4..efe1e23 100644 --- a/packages/v1-ready/salesforce/README.md +++ b/packages/v1-ready/salesforce/README.md @@ -43,4 +43,20 @@ The integration user needs additional permissions depending on which Salesforce | **Campaigns, CampaignMembers** | Enable the **Marketing User** checkbox on the user record (**Setup → Users → [user] → Edit → Advanced User Details → Marketing User ✓**). This is a feature gate separate from object-level CRUD — even a user with full Edit on Campaigns cannot insert a Campaign without it. | | **Contacts, Accounts, Leads, Opportunities** | Standard API access is sufficient. No additional flags needed beyond the base integration user profile. | -> **Open question — license compatibility:** The Marketing User flag historically consumes a Marketing User feature license. It is currently unconfirmed whether this flag can be assigned to a Salesforce Integration User license (which is stripped-down and cheaper). If your integration creates Campaigns and you plan to use an Integration license, verify this in a Developer org before advising customers — it may require a full Salesforce seat instead. \ No newline at end of file +> **Open question — license compatibility:** The Marketing User flag historically consumes a Marketing User feature license. It is currently unconfirmed whether this flag can be assigned to a Salesforce Integration User license (which is stripped-down and cheaper). If your integration creates Campaigns and you plan to use an Integration license, verify this in a Developer org before advising customers — it may require a full Salesforce seat instead. + +## Rate limits + +Salesforce counts the API calls of an org over a rolling 24 hours ([API request limits and allocations](https://developer.salesforce.com/docs/platform/salesforce-app-limits-cheatsheet/guide/salesforce-app-limits-platform-api.html)). An org over its limit gets a `403` with the error code `REQUEST_LIMIT_EXCEEDED`. Salesforce documents no reset time for it. + +`Api.rateLimit` declares that limit for the Frigg Requester. The module drives jsforce, not the Requester, so `api.withLimits(call)` applies the declaration to jsforce calls. It awaits `call()`. When the call fails with `REQUEST_LIMIT_EXCEEDED`, it throws a `RateLimitError` with `reason: 'daily'` and a `retryAt` one hour ahead, because there is no reset time to wait for and the module probes every hour. `cause` keeps the jsforce error. Any other error is rethrown unchanged. + +The data methods (`create`, `update`, `upsert`, `list`, `find`, `get`, `delete`, `getGlobalMetadata`) use `withLimits` already. `api.conn` is a plain jsforce connection, so a consumer that calls it directly wraps its own calls: + +```javascript +const result = await api.withLimits(() => api.conn.query(soql)); +``` + +Salesforce also returns `REQUEST_LIMIT_EXCEEDED` when too many requests that run for 20 seconds or longer are in flight at once (25 in a production org, 5 in a Developer Edition org). That limit clears in seconds, not in a day. Salesforce documents no field that tells the two apart, so `withLimits` reads the error message: a message that contains "concurrent" (Salesforce sends `ConcurrentPerOrgLongTxn` for it) is reported as `reason: 'concurrency'` with a `retryAt` 30 seconds ahead. Any other message stays a daily limit. + +When it throws a `RateLimitError` for a limit that lasts longer than the in-process cap (5 minutes by default: the daily limit, not the 30 second concurrency limit), `withLimits` also has the integration record one rate-limit warning for its users, as the Frigg Requester does. On a core that cannot record the warning, it only throws. A core that does not export `RateLimitError` and `classifyRateLimit` (older than the release that added them) makes `withLimits` rethrow the original jsforce error. diff --git a/packages/v1-ready/salesforce/api.js b/packages/v1-ready/salesforce/api.js index fdfb64e..ec4f33f 100644 --- a/packages/v1-ready/salesforce/api.js +++ b/packages/v1-ready/salesforce/api.js @@ -1,7 +1,18 @@ -const { get, OAuth2Requester } = require('@friggframework/core'); +const { + get, + OAuth2Requester, + RateLimitError, + classifyRateLimit, +} = require('@friggframework/core'); const jsforce = require('jsforce'); const crypto = require('crypto'); +const STATIC_WAIT_MS = { daily: 3_600_000, concurrency: 30_000 }; +const LIMITS_LINK = { + label: 'API request limits and allocations', + url: 'https://developer.salesforce.com/docs/platform/salesforce-app-limits-cheatsheet/guide/salesforce-app-limits-platform-api.html', +}; + class Api extends OAuth2Requester { // URL-unreserved and outside the base64url alphabet. static STATE_VERIFIER_DELIMITER = '~'; @@ -15,6 +26,22 @@ class Api extends OAuth2Requester { 'inactive_org', ]); + static rateLimit = { + scope: 'entity', + windows: [{ name: 'daily', rollingMs: 86_400_000 }], + classify({ body }) { + if (body?.errorCode !== 'REQUEST_LIMIT_EXCEEDED') return null; + const reason = /concurrent/i.test(body.message ?? '') + ? 'concurrency' + : 'daily'; + return { reason, waitMs: STATIC_WAIT_MS[reason], source: 'static' }; + }, + userHints: { + daily: { links: [LIMITS_LINK] }, + concurrency: { links: [LIMITS_LINK] }, + }, + }; + constructor(params) { super(params); this.jsforce = jsforce; @@ -156,23 +183,66 @@ class Api extends OAuth2Requester { return this.get('User', this.conn.userInfo.id); } + async withLimits(call) { + try { + return await call(); + } catch (err) { + const rateLimitError = this._rateLimitErrorFor(err); + if (!rateLimitError) throw err; + await this._notifyRateLimited?.(rateLimitError); + throw rateLimitError; + } + } + + _rateLimitErrorFor(err) { + if ( + typeof classifyRateLimit !== 'function' || + typeof RateLimitError !== 'function' + ) { + return null; + } + const hint = classifyRateLimit(this.constructor.rateLimit, { + status: err?.statusCode, + headers: {}, + body: { + errorCode: err?.errorCode ?? err?.name, + message: err?.message, + }, + }); + if (!hint) return null; + return new RateLimitError({ + hint, + module: this._telemetryModuleLabel(), + resource: this.conn.instanceUrl, + cause: err, + }); + } + async create(object, data) { - const response = await this.conn.sobject(object).create(data); + const response = await this.withLimits(() => + this.conn.sobject(object).create(data) + ); return response; } async update(object, data) { - const response = await this.conn.sobject(object).update(data); + const response = await this.withLimits(() => + this.conn.sobject(object).update(data) + ); return response; } async upsert(object, data) { - const response = await this.conn.sobject(object).upsert(data); + const response = await this.withLimits(() => + this.conn.sobject(object).upsert(data) + ); return response; } async list(object, ids = {}) { - const response = await this.conn.sobject(object).retrieve(ids); + const response = await this.withLimits(() => + this.conn.sobject(object).retrieve(ids) + ); return response; } @@ -182,24 +252,30 @@ class Api extends OAuth2Requester { returnFields = { '*': 1 }, options = {} ) { - const response = await this.conn - .sobject(object) - .find(findFilter, returnFields, options); + const response = await this.withLimits(() => + this.conn.sobject(object).find(findFilter, returnFields, options) + ); return response; } async getGlobalMetadata() { - const response = await this.conn.describeGlobal(); + const response = await this.withLimits(() => + this.conn.describeGlobal() + ); return response; } async get(object, id) { - const response = await this.conn.sobject(object).retrieve(id); + const response = await this.withLimits(() => + this.conn.sobject(object).retrieve(id) + ); return response; } async delete(object, data) { - const response = await this.conn.sobject(object).del(data); + const response = await this.withLimits(() => + this.conn.sobject(object).del(data) + ); return response; } diff --git a/packages/v1-ready/salesforce/test/api.test.js b/packages/v1-ready/salesforce/test/api.test.js index eedccd3..59954cc 100644 --- a/packages/v1-ready/salesforce/test/api.test.js +++ b/packages/v1-ready/salesforce/test/api.test.js @@ -26,6 +26,7 @@ jest.mock('jsforce', () => { }; }); +const { RateLimitError } = require('@friggframework/core'); const { Api } = require('../api'); const baseParams = { @@ -561,3 +562,296 @@ describe('Salesforce Api token refresh', () => { expect(callback).toHaveBeenCalledWith(expect.any(Error)); }); }); + +describe('Salesforce Api rate limits', () => { + const ONE_HOUR_MS = 3_600_000; + + const jsforceError = (errorCode, message) => + Object.assign(new Error(message), { name: errorCode, errorCode }); + const limitError = () => + jsforceError('REQUEST_LIMIT_EXCEEDED', 'TotalRequests Limit exceeded.'); + const invalidField = () => + jsforceError('INVALID_FIELD', "No such column 'Nope__c' on Contact"); + const rejectionOf = (promise) => promise.catch((error) => error); + + const dataMethods = [ + ['create', (api) => api.create('Contact', { LastName: 'Doe' })], + [ + 'update', + (api) => api.update('Contact', { Id: '003', LastName: 'D' }), + ], + ['upsert', (api) => api.upsert('Contact', { LastName: 'Doe' })], + ['list', (api) => api.list('Contact', ['003'])], + ['find', (api) => api.find('Contact', { LastName: 'Doe' })], + ['getGlobalMetadata', (api) => api.getGlobalMetadata()], + ['get', (api) => api.get('Contact', '003')], + ['delete', (api) => api.delete('Contact', '003')], + ]; + + let api; + let sobject; + + beforeEach(() => { + jest.clearAllMocks(); + api = new Api({ + ...baseParams, + delegate: { ...baseParams.delegate, name: 'salesforce' }, + }); + sobject = { + create: jest.fn(), + update: jest.fn(), + upsert: jest.fn(), + retrieve: jest.fn(), + find: jest.fn(), + del: jest.fn(), + }; + api.conn.sobject = jest.fn(() => sobject); + api.conn.describeGlobal = jest.fn(); + }); + + afterEach(() => { + jest.restoreAllMocks(); + jest.dontMock('@friggframework/core'); + delete api.conn.sobject; + delete api.conn.describeGlobal; + }); + + describe('withLimits', () => { + const now = Date.parse('2026-09-28T12:00:00Z'); + + beforeEach(() => { + jest.spyOn(Date, 'now').mockReturnValue(now); + }); + + it('returns what the call returns', async () => { + const result = await api.withLimits(async () => ({ totalSize: 1 })); + + expect(result).toEqual({ totalSize: 1 }); + }); + + it('turns REQUEST_LIMIT_EXCEEDED into a RateLimitError for a daily limit', async () => { + const cause = limitError(); + + const error = await rejectionOf( + api.withLimits(() => Promise.reject(cause)) + ); + + expect(error).toBeInstanceOf(RateLimitError); + expect(error).toMatchObject({ + isRateLimited: true, + reason: 'daily', + source: 'static', + waitMs: ONE_HOUR_MS, + module: 'salesforce', + }); + expect(error.cause).toBe(cause); + expect(error.retryAt).toEqual(new Date(now + ONE_HOUR_MS)); + }); + + it('names the org instance in the error', async () => { + const error = await rejectionOf( + api.withLimits(() => Promise.reject(limitError())) + ); + + expect(error.url).toBe(api.conn.instanceUrl); + }); + + it.each([ + ['INVALID_FIELD', invalidField], + [ + 'INSUFFICIENT_ACCESS', + () => jsforceError('INSUFFICIENT_ACCESS', 'No access'), + ], + ['no error code', () => new Error('socket hang up')], + ])('rethrows %s unchanged', async (_label, build) => { + const original = build(); + + const error = await rejectionOf( + api.withLimits(() => Promise.reject(original)) + ); + + expect(error).toBe(original); + }); + + it('tells the delegate about the limit before it throws', async () => { + const notify = jest.fn().mockResolvedValue(undefined); + api._notifyRateLimited = notify; + + const error = await rejectionOf( + api.withLimits(() => Promise.reject(limitError())) + ); + + expect(error).toBeInstanceOf(RateLimitError); + expect(notify).toHaveBeenCalledTimes(1); + expect(notify).toHaveBeenCalledWith(error); + }); + + it('does not tell the delegate about an error that is not a limit', async () => { + const notify = jest.fn().mockResolvedValue(undefined); + api._notifyRateLimited = notify; + + await rejectionOf( + api.withLimits(() => Promise.reject(invalidField())) + ); + + expect(notify).not.toHaveBeenCalled(); + }); + + it('still throws the RateLimitError on a core that cannot notify', async () => { + api._notifyRateLimited = undefined; + + const error = await rejectionOf( + api.withLimits(() => Promise.reject(limitError())) + ); + + expect(error).toBeInstanceOf(RateLimitError); + }); + + it.each([ + 'ConcurrentPerOrgLongTxn Concurrent API Request Limit exceeded', + 'Concurrent requests limit exceeded.', + ])( + 'reports %j as a concurrency limit, which clears in seconds', + async (message) => { + const error = await rejectionOf( + api.withLimits(() => + Promise.reject( + jsforceError('REQUEST_LIMIT_EXCEEDED', message) + ) + ) + ); + + expect(error).toMatchObject({ + reason: 'concurrency', + source: 'static', + waitMs: 30_000, + }); + expect(error.retryAt).toEqual(new Date(now + 30_000)); + } + ); + + it('reports any other REQUEST_LIMIT_EXCEEDED message as a daily limit', async () => { + const error = await rejectionOf( + api.withLimits(() => + Promise.reject( + jsforceError('REQUEST_LIMIT_EXCEEDED', 'Something new') + ) + ) + ); + + expect(error).toMatchObject({ + reason: 'daily', + waitMs: ONE_HOUR_MS, + }); + }); + + it('links the concurrency reason to the Salesforce API request limits too', () => { + expect(Api.rateLimit.userHints.concurrency.links).toEqual( + Api.rateLimit.userHints.daily.links + ); + }); + + it('rethrows the original error on a core without the rate-limit exports', async () => { + let OldCoreApi; + jest.isolateModules(() => { + jest.doMock('@friggframework/core', () => { + const { + RateLimitError: _error, + classifyRateLimit: _classify, + ...rest + } = jest.requireActual('@friggframework/core'); + return rest; + }); + OldCoreApi = require('../api').Api; + }); + const oldApi = new OldCoreApi({ + ...baseParams, + delegate: { ...baseParams.delegate, name: 'salesforce' }, + }); + const cause = limitError(); + + const error = await rejectionOf( + oldApi.withLimits(() => Promise.reject(cause)) + ); + + expect(error).toBe(cause); + }); + + it('links the daily reason to the Salesforce API request limits', () => { + expect(Api.rateLimit.userHints.daily.links).toEqual([ + { + label: expect.any(String), + url: 'https://developer.salesforce.com/docs/platform/salesforce-app-limits-cheatsheet/guide/salesforce-app-limits-platform-api.html', + }, + ]); + }); + }); + + describe('the data methods', () => { + it.each(dataMethods)( + '%s raises a RateLimitError when Salesforce reports REQUEST_LIMIT_EXCEEDED', + async (_name, invoke) => { + const cause = limitError(); + Object.values(sobject).forEach((fn) => + fn.mockRejectedValue(cause) + ); + api.conn.describeGlobal.mockRejectedValue(cause); + + const error = await rejectionOf(invoke(api)); + + expect(error).toBeInstanceOf(RateLimitError); + expect(error.cause).toBe(cause); + } + ); + + it.each(dataMethods)( + '%s rethrows any other Salesforce error unchanged', + async (_name, invoke) => { + const original = invalidField(); + Object.values(sobject).forEach((fn) => + fn.mockRejectedValue(original) + ); + api.conn.describeGlobal.mockRejectedValue(original); + + const error = await rejectionOf(invoke(api)); + + expect(error).toBe(original); + } + ); + + it('create passes its arguments to jsforce and returns the result', async () => { + sobject.create.mockResolvedValue({ id: '003', success: true }); + + const result = await api.create('Contact', { LastName: 'Doe' }); + + expect(api.conn.sobject).toHaveBeenCalledWith('Contact'); + expect(sobject.create).toHaveBeenCalledWith({ LastName: 'Doe' }); + expect(result).toEqual({ id: '003', success: true }); + }); + + it('find passes its arguments to jsforce and returns the result', async () => { + sobject.find.mockResolvedValue([{ Id: '003' }]); + + const result = await api.find( + 'Contact', + { LastName: 'Doe' }, + { Id: 1 }, + { limit: 5 } + ); + + expect(sobject.find).toHaveBeenCalledWith( + { LastName: 'Doe' }, + { Id: 1 }, + { limit: 5 } + ); + expect(result).toEqual([{ Id: '003' }]); + }); + }); + + it('declares a rolling 24 hour limit per org', () => { + expect(Api.rateLimit).toMatchObject({ + scope: 'entity', + windows: [{ name: 'daily', rollingMs: 86_400_000 }], + }); + }); +}); From 1b8f32580af5ed7b34452ceb1cb60dfdbb18e429 Mon Sep 17 00:00:00 2001 From: d-klotz Date: Tue, 29 Sep 2026 19:10:36 -0300 Subject: [PATCH 3/3] feat(pipedrive): declare the rate-limit policy Pipedrive limits an OAuth app per token over a rolling 2 second window and sends `x-ratelimit-reset` with the throttled response. A recorded 429 also carries `Retry-After`. The Requester met every such 429 with its fixed 1, 3, 10, 30, 60 and 180 second ladder. Declare `static rateLimit`: - `scope: 'entity'` and a burst window of 80 requests per 2 seconds. That is the lowest OAuth allowance in the docs (Lite 80, Growth 160, Premium 400, Ultimate 480). Pipedrive reports the real figure in `x-ratelimit-limit`. - `parsers: ['retryAfter', 'resetHeaders']`, the two header families Pipedrive sends. `x-ratelimit-reset` is a wait in seconds. - No `classify`. A spent daily token budget answers with a 429 and no documented wait time. Unless the response carries one of those headers, it keeps the fixed ladder and then the redelivery it has today. Add `RateLimitPolicy` and the types it uses to the module's `frigg-core.d.ts` shim, copied from the core types (friggframework/frigg#656). The subclass declares its own static, so the shim's `OAuth2Requester` needs no change. Core reads `x-ratelimit-reset` by default, so the wait tests characterise the behaviour with recorded headers, and the declaration test pins the numbers. The tests need a `@friggframework/core` with rate-limit awareness. The dependency range is unchanged until the release that ships it is known. Refs ADR-049. Co-Authored-By: Claude Sonnet 5.5 Co-Authored-By: Claude Opus 5.5 --- packages/v1-ready/pipedrive/README.md | 8 ++ packages/v1-ready/pipedrive/frigg-core.d.ts | 48 ++++++++ packages/v1-ready/pipedrive/src/api.ts | 13 +- .../v1-ready/pipedrive/test/rateLimit.test.ts | 111 ++++++++++++++++++ 4 files changed, 179 insertions(+), 1 deletion(-) create mode 100644 packages/v1-ready/pipedrive/test/rateLimit.test.ts diff --git a/packages/v1-ready/pipedrive/README.md b/packages/v1-ready/pipedrive/README.md index 6ce9a08..881d2b0 100644 --- a/packages/v1-ready/pipedrive/README.md +++ b/packages/v1-ready/pipedrive/README.md @@ -3,3 +3,11 @@ This is the API Module for pipedrive that allows the [Frigg](https://friggframework.org) code to talk to the pipedrive API. Read more on the [Frigg documentation site](https://docs.friggframework.org/api-modules/list/pipedrive + +## Rate limits + +`Api.rateLimit` tells the Frigg Requester how Pipedrive limits calls. The numbers come from [Pipedrive's rate limiting guide](https://pipedrive.readme.io/docs/core-api-concepts-rate-limiting). + +- **Burst:** a rolling 2 second window, counted per token. An OAuth app gets 80 requests in the window on the Lite plan, 160 on Growth, 400 on Premium and 480 on Ultimate. The module declares 80, the lowest. Pipedrive reports the real figure in `x-ratelimit-limit`. +- **A throttled call** returns `429`. The Requester waits for `Retry-After` when the response has one, and otherwise for `x-ratelimit-reset` (in seconds), then retries. +- **Daily token budget:** each company has a budget of 30,000 tokens times a plan multiplier times the number of seats. It resets at midnight in Pipedrive's server time zone. When it is spent, every call gets a `429`. Pipedrive documents no wait time for it, so the module declares none. Unless the response carries `Retry-After` or `x-ratelimit-reset`, such a `429` keeps the fixed backoff ladder, and then the queue redelivers the message. diff --git a/packages/v1-ready/pipedrive/frigg-core.d.ts b/packages/v1-ready/pipedrive/frigg-core.d.ts index 013c9f9..f08aaef 100644 --- a/packages/v1-ready/pipedrive/frigg-core.d.ts +++ b/packages/v1-ready/pipedrive/frigg-core.d.ts @@ -39,6 +39,54 @@ declare module '@friggframework/core' { headers?: Record; } + export type RateLimitScope = + | "credential" + | "entity" + | "app" + | ((requester: OAuth2Requester) => 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 RateLimitHint = { + retryAt: Date; + waitMs: number; + reason: "burst" | "daily" | "monthly" | "concurrency" | "unknown"; + policy?: string; + remaining?: number; + source: "header" | "body" | "static" | "backoff"; + }; + + export type RateLimitPolicy = { + scope?: RateLimitScope; + windows?: RateLimitWindow[]; + maxConcurrency?: number; + minRetryAfterMs?: number; + maxInProcessWaitMs?: 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 OAuth2Requester { access_token: string | null; refresh_token: string | null; diff --git a/packages/v1-ready/pipedrive/src/api.ts b/packages/v1-ready/pipedrive/src/api.ts index 163b754..b1c22c8 100644 --- a/packages/v1-ready/pipedrive/src/api.ts +++ b/packages/v1-ready/pipedrive/src/api.ts @@ -1,4 +1,9 @@ -import { get, OAuth2Requester, RequestOptions } from "@friggframework/core"; +import { + get, + OAuth2Requester, + RateLimitPolicy, + RequestOptions, +} from "@friggframework/core"; import { OAuth2RequesterOptions, ActivityParams, @@ -23,6 +28,12 @@ import { } from "./types"; export class Api extends OAuth2Requester { + static rateLimit: RateLimitPolicy = { + scope: "entity", + windows: [{ name: "burst", limit: 80, perMs: 2_000 }], + parsers: ["retryAfter", "resetHeaders"], + }; + companyDomain: string | null; URLs: { activities: string; diff --git a/packages/v1-ready/pipedrive/test/rateLimit.test.ts b/packages/v1-ready/pipedrive/test/rateLimit.test.ts new file mode 100644 index 0000000..e2ff276 --- /dev/null +++ b/packages/v1-ready/pipedrive/test/rateLimit.test.ts @@ -0,0 +1,111 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { Api } from "../src/api"; + +type FakeResponse = { + status: number; + headers?: Record; + body?: unknown; +}; + +function throttledFetch(responses: FakeResponse[]) { + const queue = [...responses]; + return vi.fn(async () => { + const next = queue.shift(); + if (!next) throw new Error("unexpected fetch call"); + const body = next.body ?? {}; + return { + status: next.status, + bodyUsed: false, + headers: new Map( + Object.entries({ "Content-Type": "application/json", ...next.headers }) + ), + json: async () => body, + text: vi.fn(async () => JSON.stringify(body)), + }; + }); +} + +const ok: FakeResponse = { status: 200, body: { success: true, data: [] } }; +const limited = (headers?: Record): FakeResponse => ({ + status: 429, + headers, +}); + +const recordedBurstHeaders = { + "x-ratelimit-limit": "10", + "x-ratelimit-remaining": "0", + "x-ratelimit-reset": "10", + "Retry-After": "10", +}; + +describe("Pipedrive rate limits", () => { + let delays: Array; + + beforeEach(() => { + delays = []; + const realSetTimeout = globalThis.setTimeout; + vi.spyOn(globalThis, "setTimeout").mockImplementation((( + fn: () => void, + delay?: number, + ...args: unknown[] + ) => { + delays.push(delay); + return realSetTimeout(fn, 0, ...args); + }) as typeof setTimeout); + vi.spyOn(Math, "random").mockReturnValue(0); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + const makeApi = (fetch: ReturnType) => + new Api({ + companyDomain: "https://acme.pipedrive.com", + fetch, + requestTimeoutMs: 0, + } as any); + + it("waits as long as x-ratelimit-reset says, then retries", async () => { + const fetch = throttledFetch([ + limited({ + "x-ratelimit-limit": "80", + "x-ratelimit-remaining": "0", + "x-ratelimit-reset": "2", + }), + ok, + ]); + + const result = await makeApi(fetch).listDeals(); + + expect(result).toEqual({ success: true, data: [] }); + expect(delays).toEqual([2_000]); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it("waits the 10 seconds of a recorded 429 that carries Retry-After and x-ratelimit-reset", async () => { + const fetch = throttledFetch([limited(recordedBurstHeaders), ok]); + + await makeApi(fetch).listDeals(); + + expect(delays).toEqual([10_000]); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it("keeps the fixed backoff ladder for a 429 with no headers, as for an exhausted daily budget", async () => { + const fetch = throttledFetch([limited(), ok]); + + await makeApi(fetch).listDeals(); + + expect(delays).toEqual([1_000]); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it("declares the lowest documented OAuth burst allowance", () => { + expect(Api.rateLimit).toMatchObject({ + scope: "entity", + windows: [{ name: "burst", limit: 80, perMs: 2_000 }], + parsers: ["retryAfter", "resetHeaders"], + }); + }); +});