diff --git a/.claude/skills/frigg-scheduled-jobs/SKILL.md b/.claude/skills/frigg-scheduled-jobs/SKILL.md index 7d5a736c7..47ebe5d84 100644 --- a/.claude/skills/frigg-scheduled-jobs/SKILL.md +++ b/.claude/skills/frigg-scheduled-jobs/SKILL.md @@ -1,6 +1,6 @@ --- name: frigg-scheduled-jobs -description: "Scheduling one-time deferred jobs in a Frigg integration via AWS EventBridge Scheduler — e.g. webhook subscription renewals or delayed tasks. Covers the scheduler command API (createSchedulerCommands: scheduleJob, deleteJob, getJobStatus), how the scheduler infrastructure is auto-provisioned (enabled by scheduler.enable or any integration with webhooks), the production EventBridge vs local mock provider, auto-cleanup after execution, and the required environment variables. Use when an integration needs to run a job at a future time, renew a webhook subscription, or schedule a deferred action." +description: "Scheduling one-time deferred jobs in a Frigg integration via AWS EventBridge Scheduler — e.g. webhook subscription renewals or delayed tasks. Covers the scheduler command API (createSchedulerCommands: scheduleJob, deleteJob, getJobStatus), how the scheduler infrastructure is auto-provisioned (enabled by scheduler.enable, any integration with webhooks, or any API module that declares static rateLimit), the production EventBridge vs local mock provider, auto-cleanup after execution, and the required environment variables. Use when an integration needs to run a job at a future time, renew a webhook subscription, or schedule a deferred action." --- # Frigg Scheduled Jobs @@ -56,9 +56,10 @@ STAGE=local ## Infrastructure provisioning -The scheduler infrastructure is generated by the scheduler domain builder and is **auto-enabled** when either: +The scheduler infrastructure is generated by the scheduler domain builder and is **auto-enabled** when any of these is true: -- `appDefinition.scheduler.enable === true`, or -- any integration declares webhooks (`Definition.webhooks.enabled === true` or `Definition.webhooks === true`) — webhooks commonly need renewal scheduling. +- `appDefinition.scheduler.enable === true`, +- any integration declares webhooks (`Definition.webhooks.enabled === true` or `Definition.webhooks === true`) — webhooks commonly need renewal scheduling, or +- any API module of an integration declares `static rateLimit` — the queue worker schedules a rate-limited message that must wait more than 15 minutes. When enabled, the builder creates an `AWS::Scheduler::ScheduleGroup` (`FriggScheduleGroup`) and an IAM role for EventBridge → SQS, and exposes `SCHEDULER_ROLE_ARN` to the functions. For the builder's place in the wider infra pipeline, see the `frigg` skill's `references/infrastructure.md`. Webhook-renewal scheduling pairs with the webhook patterns in the `frigg-extensions` skill. diff --git a/.claude/skills/frigg/references/infrastructure.md b/.claude/skills/frigg/references/infrastructure.md index a11714c3d..2e9a989da 100644 --- a/.claude/skills/frigg/references/infrastructure.md +++ b/.claude/skills/frigg/references/infrastructure.md @@ -50,7 +50,7 @@ Each builder implements: ## Scheduler -The scheduler builder (`scheduler/scheduler-builder.js`) auto-enables when any integration has `webhooks.enabled = true` (or explicitly via `appDefinition.scheduler.enable = true`). It creates an EventBridge Scheduler `ScheduleGroup` + an IAM role for EventBridge → SQS and exposes `SCHEDULER_ROLE_ARN`, enabling one-time scheduled jobs (e.g., webhook subscription renewals). +The scheduler builder (`scheduler/scheduler-builder.js`) auto-enables when any integration has `webhooks.enabled = true`, or when any API module declares `static rateLimit` (or explicitly via `appDefinition.scheduler.enable = true`). It creates an EventBridge Scheduler `ScheduleGroup` + an IAM role for EventBridge → SQS and exposes `SCHEDULER_ROLE_ARN`, enabling one-time scheduled jobs (e.g., webhook subscription renewals, or a rate-limited queue message that must wait more than 15 minutes). For scheduling jobs from integration code (`createSchedulerCommands`: `scheduleJob` / `deleteJob` / `getJobStatus`), the production-vs-mock provider, auto-cleanup, and env vars, see the **frigg-scheduled-jobs** skill. diff --git a/docs/guides/INTEGRATION-PATTERNS.md b/docs/guides/INTEGRATION-PATTERNS.md index d1a0e102f..f9650b838 100644 --- a/docs/guides/INTEGRATION-PATTERNS.md +++ b/docs/guides/INTEGRATION-PATTERNS.md @@ -397,6 +397,39 @@ only when SQS will move the message to the DLQ if this attempt fails, and `false` whenever that is unknown. Use it to end a run or count lost work on the final try instead of leaving the run in progress. +### Rate-Limit Deferral + +When a provider throttles a queue handler for longer than the Requester sleeps +in process, the Requester throws `RateLimitError` with `retryAt`. Let it reach +the queue worker. Core puts the message back so it runs at `retryAt`: it sends +the body again with a delay (up to 15 minutes), or schedules it, or extends its +visibility timeout (12 hours less 20 minutes at most). The delay and the +schedule send a new message, so they do not use one of the message's three +receives. The visibility timeout and a cap (see below) do use one. +With a `processId` in the message, `Process.context.rateLimit` says the run is +waiting and until when. It holds the latest known wait of the process, and is +cleared when a message runs after that wait is over. + +Check `delivery.isLastAttempt` first, then rethrow: + +```javascript +async processBatch({ data, delivery }) { + try { + await this.syncPage(data); + } catch (error) { + if (delivery?.isLastAttempt) return this.failRun(data.processId, error); + throw error; // a RateLimitError too: core puts the message back at retryAt + } +} +``` + +A deferral past `FRIGG_QUEUE_MAX_DEFERRALS` (default 30) or +`FRIGG_QUEUE_MAX_DEFERRED_MS` (default 26 hours) fails the record like any other +retryable error, and `delivery` applies. So does a `RateLimitError` with no +`retryAt`: a non-429 limit that `classify` names without a time. The defaults cover one full day of +hourly waits, for a module that probes a daily limit every hour. Do not catch a +`RateLimitError` to write a retry of your own. + --- ## Sync Orchestration @@ -802,9 +835,9 @@ Put what you know about a provider's limits in its API module: declare 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. +and ends the run on the last attempt. Otherwise it rethrows: the queue worker +puts the message back to run at `retryAt` (see [Rate-Limit Deferral](#rate-limit-deferral)). +Do not catch it to write a retry of your own. For pacing in your own loop, respect API rate limits: diff --git a/docs/guides/LOGGING-IN-INTEGRATIONS.md b/docs/guides/LOGGING-IN-INTEGRATIONS.md index 6b65771b5..e9345dcc6 100644 --- a/docs/guides/LOGGING-IN-INTEGRATIONS.md +++ b/docs/guides/LOGGING-IN-INTEGRATIONS.md @@ -202,6 +202,16 @@ try { the last retry, the DLQ processor logs `frigg.queue.dlq.message_failed` at `ERROR`. A `HaltError` stops the retries and logs `frigg.worker.record_halted` at `ERROR`. +- A `RateLimitError` (an error with `isRateLimited`) is different. When it + has a `retryAt`, the `Worker` puts the message back to run at that time. It + logs one record: + - `frigg.worker.record_deferred` at `WARN` when it sent the message again + with a delay, or scheduled it. The new message starts with all its retries. + - One `WARN` when the message returns to SQS: + `frigg.worker.record_deferral_capped`, `record_visibility_extended`, or + `record_failed` with `deferral.skipped`. + - On the last attempt, one `ERROR`, `frigg.worker.record_lost_rate_limited`, + with the details, and no `WARN`. The message goes to the DLQ next. ## Ids that you learn later @@ -577,7 +587,8 @@ What each record carries without extra code: - The `listContacts` DEBUG records carry `entityId` and `credentialId` of the HubSpot module. - If `processBatch` throws, the `Worker` logs `frigg.worker.record_failed` with - the same ids, and the message returns to SQS. + the same ids, and the message returns to SQS. For a `RateLimitError` it logs + one of the deferral records instead (see [Errors](#errors)). ## Common mistakes diff --git a/docs/reference/api-module-definition-and-functions.md b/docs/reference/api-module-definition-and-functions.md index 23712f64d..8057371cc 100644 --- a/docs/reference/api-module-definition-and-functions.md +++ b/docs/reference/api-module-definition-and-functions.md @@ -330,8 +330,10 @@ A Requester built with `backOff: []` never sleeps: a hinted wait throws 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. +worker does not halt a `RateLimitError`, even when its status is 403. It puts +the message back to run at `retryAt` (see the Rate-Limit Deferral section of +the [Integration Patterns Guide](/docs/guides/INTEGRATION-PATTERNS.md)). With +no `retryAt`, the message returns to SQS like any retryable error. ### A client that is not the Requester diff --git a/package-lock.json b/package-lock.json index 1af7843c1..66fb19d4f 100644 --- a/package-lock.json +++ b/package-lock.json @@ -34029,6 +34029,7 @@ "@aws-sdk/client-kms": "^3.588.0", "@aws-sdk/client-lambda": "^3.714.0", "@aws-sdk/client-s3": "^3.588.0", + "@aws-sdk/client-scheduler": "^3.588.0", "@aws-sdk/client-sqs": "^3.588.0", "@aws-sdk/client-ssm": "^3.588.0", "@aws-sdk/s3-request-presigner": "^3.588.0", @@ -34063,6 +34064,7 @@ "@prisma/client": "^6.19.3", "@types/lodash": "4.17.15", "@typescript-eslint/eslint-plugin": "^8.0.0", + "aws-sdk-client-mock": "^4.1.0", "chai": "^4.3.6", "eslint": "^8.22.0", "eslint-plugin-import": "^2.29.1", diff --git a/packages/core/CLAUDE.md b/packages/core/CLAUDE.md index 5f827ca62..f950b5843 100644 --- a/packages/core/CLAUDE.md +++ b/packages/core/CLAUDE.md @@ -486,20 +486,22 @@ 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 and 429). +(`ownership.queue: 'external'`). The value does not change what core does: a +retryable error returns the message to SQS, a rate-limited error puts it back +to run at `retryAt` (see **Rate-limit deferral** below), and a halt error (4xx +except 408, 429 and errors with `isRateLimited`) discards it. -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 -`true`, the message goes to the DLQ next, so mark the run failed or count the -message's records as failed before rethrowing. +Use it to end a run or count lost work on the final try. Check it first: when +it is `true`, mark the run failed or count the message's records as failed, and +return. Otherwise rethrow. Do not end the run and then rethrow: core can still +send a `RateLimitError` again, so the message would run again on an ended run. ```javascript async processBatch({ data, delivery }) { try { await this.syncPage(data); } catch (error) { - if (delivery?.isLastAttempt) await this.failRun(data.processId, error); + if (delivery?.isLastAttempt) return this.failRun(data.processId, error); throw error; } } @@ -510,6 +512,76 @@ The single source of the max receive count is devtools integration builder uses it for the queue's `RedrivePolicy` and sets it as `FRIGG_QUEUE_MAX_RECEIVE_COUNT` on the queue worker function. +**Rate-limit deferral** (ADR-049): when a queue handler throws an error with +`isRateLimited` and a `retryAt` (a `RateLimitError` from the Requester, or one a +module throws), `Worker.run` does not give the message back to SQS for a 30 +minute redelivery. It puts the message back so it runs at `retryAt`: + +| Wait | What the worker does | Record | +|---|---|---| +| Up to 900 s | Sends the same body again with `DelaySeconds`, then acknowledges the old message | handled | +| Over 900 s, and `SCHEDULER_ROLE_ARN` is set | Creates a one-time EventBridge schedule at `retryAt` that sends the body to the queue, then acknowledges | handled | +| Over 900 s, no scheduler | Extends the message's visibility timeout to `retryAt` (12 h less 20 minutes at most). This uses one receive | reported failed | + +- The new message starts at receive count 1. A deferred body carries + `_frigg: { deferrals, firstDeferredAt, deferralId }`; `_frigg` is reserved and + the handler never sees it. `deferralId` is the SQS message id of the first + message that was put back, and every later deferral of that work keeps it. +- The visibility tier sets a timeout of at most 12 h less 20 minutes + (42,000 s). SQS counts its 12 h maximum from the receive, and an extension + does not reset it. The margin covers the longest Lambda run (900 s) plus the + longest batching window (300 s). +- Two caps bound it: `FRIGG_QUEUE_MAX_DEFERRALS` (default 30) and + `FRIGG_QUEUE_MAX_DEFERRED_MS` (default 26 h, from the first deferral to the + retry time). The defaults cover one full day of hourly waits, for a module + that probes a daily limit every hour. Past a cap, with no `retryAt`, with no + event source ARN or on a FIFO queue, the record fails like any retryable + error and `delivery` applies. +- The mock scheduler is never used for a deferral: a schedule kept in memory + would be lost with the acknowledged message. +- With a `processId` in the message, the worker writes + `Process.context.rateLimit = { status: 'WAITING' | 'EXHAUSTED', mechanism, + retryAt, reason, module, deferrals, deferralId, updatedAt }`. `EXHAUSTED` means + a cap ended the deferrals, or a visibility change came on the last delivery. + The value is one per process, so it holds the latest known wait: a new wait + replaces it only when its `retryAt` is later, and a `WAITING` value never hides + an `EXHAUSTED` one. After a deferred or redelivered message ran, the worker + sets a `WAITING` value to `null` at once when that message set it (same + `deferralId`, or the message's own id for a visibility change), and for any + other message only when its `retryAt` is past or within 5 s. So a job that + resumes before its `retryAt` clears its own wait, and cannot clear the wait of + another message. +- Delivery stays at least once. The new message is sent before the old one is + acknowledged, so a crash between the two can run the work twice. +- Handlers check `delivery.isLastAttempt` first, then rethrow. The delay and + schedule tiers send a new message, so they never reach the last attempt. The + visibility tier and a cap use a receive: on the last one the worker writes an + ERROR (`frigg.worker.record_lost_rate_limited`), and the handler has this one + chance to end the run. + +```javascript +async processBatch({ data, delivery }) { + try { + await this.syncPage(data); + } catch (error) { + if (delivery?.isLastAttempt) return this.failRun(data.processId, error); + throw error; // a RateLimitError too: core puts the message back at retryAt + } +} +``` + +Records: a rate-limited SQS record gets one log record, by its outcome. + +| Outcome | Log record | +|---|---| +| Acknowledged | WARN `frigg.worker.record_deferred` (`mechanism` `delay` or `schedule`) | +| Not acknowledged, not the last attempt | One WARN: `record_deferral_capped`, `record_visibility_extended`, or `record_failed` with `deferral.skipped` | +| Not acknowledged, last attempt | One ERROR, `record_lost_rate_limited`, with the details, and no WARN | + +A failing run state hook adds `rate_limit_state_failed`. A queue in a stack +that does not own it (`ownership.queue: 'external'`) needs its own +`sqs:ChangeMessageVisibility` and scheduler access. + ### 8. Error Handling (`/errors`) **Purpose**: Standardized error types with proper HTTP semantics. @@ -761,6 +833,8 @@ Use test doubles from `@friggframework/test` package for consistent mocking. - `SECRET_ARN` - AWS Secrets Manager ARN for auto-injection - `FRIGG_LOG_LEVEL` - Minimum log level (`TRACE` … `FATAL`, default `INFO`; `DEBUG` on local runs) - `FRIGG_QUEUE_MAX_RECEIVE_COUNT` - Set by devtools on integration queue workers whose queue the stack owns; feeds `delivery.maxReceiveCount` +- `FRIGG_QUEUE_MAX_DEFERRALS` - Most rate-limit deferrals of one message (default 30) +- `FRIGG_QUEUE_MAX_DEFERRED_MS` - Most time in ms from a message's first rate-limit deferral to its retry time (default 26 h) ## Version Information diff --git a/packages/core/core/CLAUDE.md b/packages/core/core/CLAUDE.md index 905de2de5..3ef4fbab3 100644 --- a/packages/core/core/CLAUDE.md +++ b/packages/core/core/CLAUDE.md @@ -79,6 +79,36 @@ await worker.send({ }, delaySeconds); ``` +**Rate-limit deferral** (ADR-049): `run()` handles an error with `isRateLimited` +and a `retryAt` through `defer(record, body, error, delivery)`, which returns +`{ outcome }`: +- `acked`: the message was sent again with `DelaySeconds` (wait up to 900 s) or + scheduled with the one-time scheduler (longer wait). The record is handled. +- `failed`: the visibility timeout was extended to `retryAt`, or a cap + (`FRIGG_QUEUE_MAX_DEFERRALS`, `FRIGG_QUEUE_MAX_DEFERRED_MS`) ended the + deferrals. The record is reported failed. +- `skipped`: no `retryAt` (`no_retry_at`), no `eventSourceARN` or a FIFO queue + (`no_event_source`), a failed send (`send_failed`), or a failed visibility + change (`visibility_failed`). The record fails as any error does. + +Each outcome writes one log record. `acked` writes WARN `record_deferred`. +`failed` and `skipped` write one WARN (`record_visibility_extended`, +`record_deferral_capped`, or `record_failed` with `deferral: { skipped }`). On +the last attempt they write one ERROR, `record_lost_rate_limited`, with the +details, and no WARN. + +Subclasses override two no-op hooks to keep run state. +`recordRateLimitWait(body, error, state)` runs before a delayed message is sent, +because with a short delay the new message can run first; a failed send clears +the wait again. For a schedule or a visibility change it runs after the change. +`clearRateLimitWait(body, { deferralId })` runs after a deferred or redelivered +message succeeded. +`state.deferralId` and `deferralId` name the message that was first put back +(`_frigg.deferralId`, or the message's own id). A failing hook logs +`frigg.worker.rate_limit_state_failed` and never changes the outcome. The queue +worker of an integration (`createQueueWorker`) writes `Process.context.rateLimit` +in them. See `packages/core/CLAUDE.md` section 7. + ### Delegate Pattern System (`Delegate.js:3-27`) **Purpose**: Observer/delegation pattern for decoupled component communication diff --git a/packages/core/core/Worker.deferral.test.js b/packages/core/core/Worker.deferral.test.js new file mode 100644 index 000000000..b755b4ea9 --- /dev/null +++ b/packages/core/core/Worker.deferral.test.js @@ -0,0 +1,891 @@ +const { mockClient } = require('aws-sdk-client-mock'); +const { + ChangeMessageVisibilityCommand, + GetQueueUrlCommand, + SQSClient, + SendMessageCommand, +} = require('@aws-sdk/client-sqs'); +const { + CreateScheduleCommand, + SchedulerClient, +} = require('@aws-sdk/client-scheduler'); +const { Worker } = require('./Worker'); +const { RateLimitError } = require('../errors'); +const { createMemorySink } = require('../logs'); +const { + MAX_DEFERRALS_ENV, + MAX_DEFERRED_MS, + MAX_DEFERRED_MS_ENV, +} = require('../queues/queue-deferral'); + +const NOW = Date.parse('2026-09-28T12:00:00.000Z'); +const SECOND = 1_000; +const MINUTE = 60 * SECOND; +const HOUR = 60 * MINUTE; + +const QUEUE_NAME = 'app--dev-ExampleQueue'; +const QUEUE_ARN = `arn:aws:sqs:us-east-1:123456789012:${QUEUE_NAME}`; +const QUEUE_URL = `https://sqs.us-east-1.amazonaws.com/123456789012/${QUEUE_NAME}`; +const SCHEDULER_ROLE_ARN = + 'arn:aws:iam::123456789012:role/app-dev-scheduler-role'; +const BODY = { event: 'SYNC_PAGE', data: { processId: 7 } }; + +const ENV_KEYS = [ + 'AWS_ENDPOINT', + 'FRIGG_QUEUE_MAX_RECEIVE_COUNT', + 'IS_OFFLINE', + 'SCHEDULE_GROUP_NAME', + 'SCHEDULER_PROVIDER', + 'SCHEDULER_ROLE_ARN', + MAX_DEFERRALS_ENV, + MAX_DEFERRED_MS_ENV, +]; + +function rateLimited(waitMs, { reason = 'burst', module = 'hubspot' } = {}) { + return new RateLimitError({ + resource: 'https://api.example.com/v1/items', + response: { status: 429, bodyUsed: true }, + hint: { waitMs, reason, source: 'header' }, + waitMs, + module, + now: NOW, + }); +} + +function sqsRecord(overrides = {}) { + return { + messageId: 'msg-1', + receiptHandle: 'receipt-1', + eventSourceARN: QUEUE_ARN, + attributes: { ApproximateReceiveCount: '1' }, + body: JSON.stringify(BODY), + ...overrides, + }; +} + +function deferredBody( + deferrals, + firstDeferredAt = new Date(NOW).toISOString(), + deferralId = 'msg-1' +) { + return { ...BODY, _frigg: { deferrals, firstDeferredAt, deferralId } }; +} + +function deferredRecord(deferrals, firstDeferredAt) { + return sqsRecord({ + body: JSON.stringify(deferredBody(deferrals, firstDeferredAt)), + }); +} + +describe('Worker rate-limit deferral (ADR-049)', () => { + let sqsMock; + let schedulerMock; + let sink; + let worker; + let savedEnv; + + beforeEach(() => { + savedEnv = {}; + for (const key of ENV_KEYS) { + savedEnv[key] = process.env[key]; + delete process.env[key]; + } + jest.spyOn(Date, 'now').mockReturnValue(NOW); + jest.spyOn(console, 'log').mockImplementation(() => {}); + jest.spyOn(console, 'warn').mockImplementation(() => {}); + sqsMock = mockClient(SQSClient); + schedulerMock = mockClient(SchedulerClient); + sqsMock.on(GetQueueUrlCommand).resolves({ QueueUrl: QUEUE_URL }); + sqsMock.on(SendMessageCommand).resolves({ MessageId: 'new-message' }); + sqsMock.on(ChangeMessageVisibilityCommand).resolves({}); + schedulerMock.on(CreateScheduleCommand).resolves({ + ScheduleArn: + 'arn:aws:scheduler:us-east-1:123456789012:schedule/g/n', + }); + sink = createMemorySink(); + worker = new Worker(); + worker._run = jest.fn(); + }); + + afterEach(() => { + jest.restoreAllMocks(); + sqsMock.restore(); + schedulerMock.restore(); + for (const key of ENV_KEYS) { + if (savedEnv[key] === undefined) delete process.env[key]; + else process.env[key] = savedEnv[key]; + } + }); + + const sent = () => + sqsMock + .commandCalls(SendMessageCommand) + .map((call) => call.args[0].input); + const sentBody = (index = 0) => JSON.parse(sent()[index].MessageBody); + const visibility = () => + sqsMock + .commandCalls(ChangeMessageVisibilityCommand) + .map((call) => call.args[0].input); + const schedules = () => + schedulerMock + .commandCalls(CreateScheduleCommand) + .map((call) => call.args[0].input); + const logged = (name) => + sink.records.filter((r) => r.eventName === `frigg.worker.${name}`); + const run = (record = sqsRecord()) => worker.run({ Records: [record] }); + const failure = (id = 'msg-1') => ({ + batchItemFailures: [{ itemIdentifier: id }], + }); + + describe('a wait of up to 15 minutes', () => { + it('puts the message back with DelaySeconds and a deferral counter, and reports it handled', async () => { + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + const result = await run(); + + expect(result).toEqual({ batchItemFailures: [] }); + expect(sent()).toHaveLength(1); + expect(sent()[0]).toMatchObject({ + QueueUrl: QUEUE_URL, + DelaySeconds: 60, + }); + expect(sentBody()).toEqual(deferredBody(1)); + expect(worker._run).toHaveBeenCalledTimes(1); + }); + + it('rounds the delay up to whole seconds', async () => { + worker._run.mockRejectedValue(rateLimited(1_500)); + await run(); + expect(sent()[0].DelaySeconds).toBe(2); + }); + + it('sends with a delay of 900 seconds for a wait of exactly 15 minutes', async () => { + worker._run.mockRejectedValue(rateLimited(15 * MINUTE)); + await run(); + expect(sent()[0].DelaySeconds).toBe(900); + expect(schedules()).toEqual([]); + }); + + it('sends with no delay when the retry time has passed', async () => { + const error = new RateLimitError({ + resource: 'https://api.example.com/v1/items', + response: { status: 429, bodyUsed: true }, + hint: { retryAt: new Date(NOW - 5 * SECOND), source: 'header' }, + now: NOW, + }); + worker._run.mockRejectedValue(error); + + await run(); + + expect(sent()[0].DelaySeconds).toBe(0); + }); + + it('marks the new message with the id of the message it replaces', async () => { + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + await run(); + + expect(sentBody()._frigg.deferralId).toBe('msg-1'); + }); + + it('keeps the id of the first message when the message is put back again', async () => { + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + await run( + sqsRecord({ + messageId: 'msg-2', + body: JSON.stringify(deferredBody(1, undefined, 'msg-1')), + }) + ); + + expect(sentBody()._frigg.deferralId).toBe('msg-1'); + }); + + it('counts up and keeps the first deferral time on a message that was deferred before', async () => { + const firstDeferredAt = new Date(NOW - 10 * MINUTE).toISOString(); + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + await run(deferredRecord(2, firstDeferredAt)); + + expect(sentBody()._frigg).toEqual({ + deferrals: 3, + firstDeferredAt, + deferralId: 'msg-1', + }); + }); + + it('looks the queue URL up from the event source ARN, once per queue', async () => { + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + await worker.run({ + Records: [sqsRecord(), sqsRecord({ messageId: 'msg-2' })], + }); + + const lookups = sqsMock.commandCalls(GetQueueUrlCommand); + expect(lookups).toHaveLength(1); + expect(lookups[0].args[0].input).toEqual({ QueueName: QUEUE_NAME }); + expect(sent()).toHaveLength(2); + }); + + it('tries the queue URL lookup again after it failed', async () => { + sqsMock + .on(GetQueueUrlCommand) + .rejectsOnce(new Error('throttled')) + .resolves({ QueueUrl: QUEUE_URL }); + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + const result = await worker.run({ + Records: [sqsRecord(), sqsRecord({ messageId: 'msg-2' })], + }); + + expect(result).toEqual(failure('msg-1')); + expect(sent()).toHaveLength(1); + }); + + it('hands the deferred message to _run with the counters and a fresh delivery', async () => { + const firstDeferredAt = new Date(NOW - MINUTE).toISOString(); + worker._run.mockResolvedValue(undefined); + + await run(deferredRecord(1, firstDeferredAt)); + + expect(worker._run).toHaveBeenCalledWith( + deferredBody(1, firstDeferredAt), + {}, + expect.objectContaining({ receiveCount: 1 }) + ); + }); + }); + + describe('a wait over 15 minutes', () => { + beforeEach(() => { + process.env.SCHEDULER_ROLE_ARN = SCHEDULER_ROLE_ARN; + }); + + it('schedules the message at the retry time with the one-time scheduler', async () => { + process.env.SCHEDULE_GROUP_NAME = 'app-dev-schedules'; + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + const result = await run(); + + expect(result).toEqual({ batchItemFailures: [] }); + expect(sent()).toEqual([]); + expect(schedules()).toHaveLength(1); + const [schedule] = schedules(); + expect(schedule).toMatchObject({ + Name: 'frigg-defer-msg-1', + GroupName: 'app-dev-schedules', + ScheduleExpression: 'at(2026-09-28T12:20:00)', + ActionAfterCompletion: 'DELETE', + Target: { Arn: QUEUE_ARN, RoleArn: SCHEDULER_ROLE_ARN }, + }); + expect(JSON.parse(schedule.Target.Input)).toEqual(deferredBody(1)); + }); + + it('takes a schedule that exists already as scheduled', async () => { + schedulerMock + .on(CreateScheduleCommand) + .rejects( + Object.assign(new Error('exists'), { + name: 'ConflictException', + }) + ); + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + const result = await run(); + + expect(result).toEqual({ batchItemFailures: [] }); + expect(visibility()).toEqual([]); + }); + + it('extends the visibility timeout when the schedule cannot be created, and reports the record failed', async () => { + schedulerMock + .on(CreateScheduleCommand) + .rejects(new Error('scheduler is down')); + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + const result = await run(); + + expect(result).toEqual(failure()); + expect(visibility()).toEqual([ + { + QueueUrl: QUEUE_URL, + ReceiptHandle: 'receipt-1', + VisibilityTimeout: 1200, + }, + ]); + expect(logged('record_visibility_extended')).toEqual([ + expect.objectContaining({ + scheduleError: expect.objectContaining({ + message: 'scheduler is down', + }), + }), + ]); + }); + + it('extends the visibility timeout when the scheduler cannot be built, and tries to build it again', async () => { + const factory = require('../infrastructure/scheduler/scheduler-service-factory'); + jest.spyOn(factory, 'createSchedulerService').mockImplementation( + () => { + throw new Error('bad scheduler config'); + } + ); + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + const first = await run(); + const second = await run(sqsRecord({ messageId: 'msg-2' })); + + expect(first).toEqual(failure()); + expect(second).toEqual(failure('msg-2')); + expect(visibility()).toHaveLength(2); + expect(logged('record_visibility_extended')).toEqual([ + expect.objectContaining({ + scheduleError: expect.objectContaining({ + message: 'bad scheduler config', + }), + }), + expect.objectContaining({ + scheduleError: expect.objectContaining({ + message: 'bad scheduler config', + }), + }), + ]); + }); + + it('never uses the mock scheduler, even when the environment names it', async () => { + process.env.SCHEDULER_PROVIDER = 'mock'; + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + const result = await run(); + + expect(result).toEqual(failure()); + expect(schedules()).toEqual([]); + expect(visibility()).toHaveLength(1); + }); + }); + + describe('a wait over 15 minutes with no scheduler', () => { + it('extends the visibility timeout to the retry time and reports the record failed', async () => { + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + const result = await run(); + + expect(result).toEqual(failure()); + expect(sent()).toEqual([]); + expect(visibility()).toEqual([ + { + QueueUrl: QUEUE_URL, + ReceiptHandle: 'receipt-1', + VisibilityTimeout: 1200, + }, + ]); + }); + + it('keeps the visibility timeout 20 minutes under the 12 hours that SQS counts from the receive', async () => { + process.env[MAX_DEFERRED_MS_ENV] = String(48 * HOUR); + worker._run.mockRejectedValue(rateLimited(13 * HOUR)); + + await run(); + + expect(visibility()[0].VisibilityTimeout).toBe(42_000); + }); + + it('leaves the message to SQS when the visibility change fails', async () => { + sqsMock + .on(ChangeMessageVisibilityCommand) + .rejects(new Error('gone')); + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + const result = await run(); + + expect(result).toEqual(failure()); + expect(logged('record_failed')).toEqual([ + expect.objectContaining({ + deferral: { + skipped: 'visibility_failed', + cause: expect.objectContaining({ message: 'gone' }), + }, + }), + ]); + }); + }); + + describe('the caps', () => { + it('reports the record failed, and sends nothing, past the deferral count', async () => { + process.env[MAX_DEFERRALS_ENV] = '2'; + const firstDeferredAt = new Date(NOW - MINUTE).toISOString(); + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + const result = await run(deferredRecord(2, firstDeferredAt)); + + expect(result).toEqual(failure()); + expect(sent()).toEqual([]); + expect(logged('record_deferral_capped')).toEqual([ + expect.objectContaining({ + level: 'WARN', + deferrals: 3, + maxDeferrals: 2, + }), + ]); + expect(logged('record_failed')).toEqual([]); + }); + + it('reports the record failed when the retry time is past the total time', async () => { + worker._run.mockRejectedValue(rateLimited(MAX_DEFERRED_MS + HOUR)); + + const result = await run(); + + expect(result).toEqual(failure()); + expect(sent()).toEqual([]); + expect(visibility()).toEqual([]); + expect(logged('record_deferral_capped')).toHaveLength(1); + }); + + it('counts the total time from the first deferral', async () => { + const firstDeferredAt = new Date( + NOW - (MAX_DEFERRED_MS - HOUR) + ).toISOString(); + worker._run.mockRejectedValue(rateLimited(2 * HOUR)); + + const result = await run(deferredRecord(1, firstDeferredAt)); + + expect(result).toEqual(failure()); + expect(logged('record_deferral_capped')).toHaveLength(1); + }); + + it('writes an ERROR when a capped message is on its last delivery', async () => { + process.env.FRIGG_QUEUE_MAX_RECEIVE_COUNT = '3'; + worker._run.mockRejectedValue(rateLimited(MAX_DEFERRED_MS + HOUR)); + + await run( + sqsRecord({ attributes: { ApproximateReceiveCount: '3' } }) + ); + + expect(logged('record_lost_rate_limited')).toEqual([ + expect.objectContaining({ + level: 'ERROR', + receiveCount: 3, + reason: 'burst', + maxDeferrals: 30, + maxDeferredMs: MAX_DEFERRED_MS, + }), + ]); + expect(logged('record_deferral_capped')).toEqual([]); + }); + + it('writes no lost-message ERROR before the last delivery', async () => { + process.env.FRIGG_QUEUE_MAX_RECEIVE_COUNT = '3'; + worker._run.mockRejectedValue(rateLimited(MAX_DEFERRED_MS + HOUR)); + + await run( + sqsRecord({ attributes: { ApproximateReceiveCount: '2' } }) + ); + + expect(logged('record_lost_rate_limited')).toEqual([]); + expect(logged('record_deferral_capped')).toHaveLength(1); + }); + + it('writes one ERROR, and no WARN, for a visibility change on the last delivery', async () => { + process.env.FRIGG_QUEUE_MAX_RECEIVE_COUNT = '3'; + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + await run( + sqsRecord({ attributes: { ApproximateReceiveCount: '3' } }) + ); + + expect(logged('record_lost_rate_limited')).toEqual([ + expect.objectContaining({ + level: 'ERROR', + mechanism: 'visibility', + visibilityTimeout: 1200, + }), + ]); + expect(logged('record_visibility_extended')).toEqual([]); + }); + + it('writes one ERROR, and no record_failed, when a deferral fails on the last delivery', async () => { + process.env.FRIGG_QUEUE_MAX_RECEIVE_COUNT = '3'; + sqsMock.on(SendMessageCommand).rejects(new Error('SQS is down')); + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + const result = await run( + sqsRecord({ attributes: { ApproximateReceiveCount: '3' } }) + ); + + expect(result).toEqual(failure()); + expect(logged('record_lost_rate_limited')).toEqual([ + expect.objectContaining({ + level: 'ERROR', + reason: 'burst', + deferral: { + skipped: 'send_failed', + cause: expect.objectContaining({ + message: 'SQS is down', + }), + }, + }), + ]); + expect(logged('record_failed')).toEqual([]); + }); + }); + + describe('a rate-limit error that cannot be deferred', () => { + it.each([ + [ + 'an error with isRateLimited and no retryAt', + () => + Object.assign(new Error('limited'), { + isRateLimited: true, + }), + ], + [ + 'a RateLimitError for a non-429 limit with no time', + () => + new RateLimitError({ + resource: 'https://api.example.com/v1/items', + response: { status: 403, bodyUsed: true }, + hint: { reason: 'daily' }, + module: 'hubspot', + now: NOW, + }), + ], + ])('goes back to SQS as before for %s', async (_label, makeError) => { + worker._run.mockRejectedValue(makeError()); + + const result = await run(); + + expect(result).toEqual(failure()); + expect(sent()).toEqual([]); + expect(schedules()).toEqual([]); + expect(visibility()).toEqual([]); + expect(logged('record_failed')).toEqual([ + expect.objectContaining({ + deferral: { skipped: 'no_retry_at' }, + }), + ]); + }); + + it.each([ + ['no event source ARN', { eventSourceARN: undefined }], + ['a FIFO queue', { eventSourceARN: `${QUEUE_ARN}.fifo` }], + ])('goes back to SQS as before with %s', async (_label, overrides) => { + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + const result = await run(sqsRecord(overrides)); + + expect(result).toEqual(failure()); + expect(sent()).toEqual([]); + expect(logged('record_failed')).toEqual([ + expect.objectContaining({ + deferral: { skipped: 'no_event_source' }, + }), + ]); + }); + + it('goes back to SQS, and is not lost, when the send fails', async () => { + sqsMock.on(SendMessageCommand).rejects(new Error('SQS is down')); + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + const result = await run(); + + expect(result).toEqual(failure()); + expect(logged('record_failed')).toEqual([ + expect.objectContaining({ + deferral: { + skipped: 'send_failed', + cause: expect.objectContaining({ + message: 'SQS is down', + }), + }, + }), + ]); + }); + + it('goes back to SQS when the message body is not JSON', async () => { + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + const result = await run(sqsRecord({ body: 'not json' })); + + expect(result).toEqual(failure()); + expect(sent()).toEqual([]); + }); + + it('does not touch an error that is not a rate limit', async () => { + worker._run.mockRejectedValue(new Error('boom')); + + const result = await run(); + + expect(result).toEqual(failure()); + expect(sent()).toEqual([]); + expect(logged('record_failed')[0].deferral).toBeUndefined(); + }); + }); + + describe('logging', () => { + it('writes one WARN record_deferred with the wait, the times and the counters', async () => { + worker._run.mockRejectedValue( + rateLimited(60 * SECOND, { reason: 'daily' }) + ); + + await run( + sqsRecord({ attributes: { ApproximateReceiveCount: '2' } }) + ); + + expect(logged('record_deferred')).toEqual([ + expect.objectContaining({ + level: 'WARN', + logger: 'frigg.worker', + messageId: 'msg-1', + receiveCount: 2, + mechanism: 'delay', + waitMs: 60_000, + delaySeconds: 60, + retryAt: new Date(NOW + 60_000).toISOString(), + deferrals: 1, + firstDeferredAt: new Date(NOW).toISOString(), + reason: 'daily', + module: 'hubspot', + statusCode: 429, + error: expect.objectContaining({ type: 'RateLimitError' }), + }), + ]); + expect(logged('record_failed')).toEqual([]); + }); + + it('names the schedule when the message was scheduled', async () => { + process.env.SCHEDULER_ROLE_ARN = SCHEDULER_ROLE_ARN; + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + await run(); + + expect(logged('record_deferred')).toEqual([ + expect.objectContaining({ + mechanism: 'schedule', + scheduleName: 'frigg-defer-msg-1', + }), + ]); + }); + + it('writes one WARN record_visibility_extended and no record_failed', async () => { + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + await run(); + + expect(logged('record_visibility_extended')).toEqual([ + expect.objectContaining({ + level: 'WARN', + mechanism: 'visibility', + visibilityTimeout: 1200, + waitMs: 20 * MINUTE, + }), + ]); + expect(logged('record_failed')).toEqual([]); + }); + }); + + describe('the run state hooks', () => { + beforeEach(() => { + worker.recordRateLimitWait = jest.fn().mockResolvedValue(undefined); + worker.clearRateLimitWait = jest.fn().mockResolvedValue(undefined); + }); + + const keepWaitInMemory = () => { + const store = { wait: null }; + worker.recordRateLimitWait.mockImplementation( + async (body, error, state) => { + store.wait = state; + } + ); + worker.clearRateLimitWait.mockImplementation( + async (body, { deferralId }) => { + if (store.wait?.deferralId === deferralId) { + store.wait = null; + } + } + ); + return store; + }; + + it('records the wait before the delayed message is sent, so a new message that runs at once clears it', async () => { + const store = keepWaitInMemory(); + sqsMock.on(SendMessageCommand).callsFake(async (input) => { + const body = JSON.parse(input.MessageBody); + await worker.clearRateLimitWait(body, { + deferralId: body._frigg.deferralId, + }); + return { MessageId: 'new-message' }; + }); + worker._run.mockRejectedValue(rateLimited(0)); + + const result = await run(); + + expect(result).toEqual({ batchItemFailures: [] }); + expect(sent()).toHaveLength(1); + expect(store.wait).toBeNull(); + }); + + it('records the wait of a delayed message', async () => { + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + await run(); + + expect(worker.recordRateLimitWait).toHaveBeenCalledTimes(1); + const [body, error, state] = + worker.recordRateLimitWait.mock.calls[0]; + expect(body).toEqual(BODY); + expect(error).toBeInstanceOf(RateLimitError); + expect(state).toEqual({ + status: 'WAITING', + mechanism: 'delay', + deferrals: 1, + deferralId: 'msg-1', + retryAt: new Date(NOW + 60_000), + }); + }); + + it('leaves no wait when the delayed message could not be sent', async () => { + const store = keepWaitInMemory(); + sqsMock.on(SendMessageCommand).rejects(new Error('SQS is down')); + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + const result = await run(); + + expect(result).toEqual(failure()); + expect(store.wait).toBeNull(); + }); + + it('records nothing when the visibility change fails', async () => { + sqsMock + .on(ChangeMessageVisibilityCommand) + .rejects(new Error('gone')); + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + await run(); + + expect(worker.recordRateLimitWait).not.toHaveBeenCalled(); + }); + + it('records EXHAUSTED when a cap ended the deferrals', async () => { + worker._run.mockRejectedValue(rateLimited(MAX_DEFERRED_MS + HOUR)); + + await run(); + + expect(worker.recordRateLimitWait.mock.calls[0][2]).toMatchObject({ + status: 'EXHAUSTED', + }); + }); + + it('records EXHAUSTED for a visibility change on the last delivery', async () => { + process.env.FRIGG_QUEUE_MAX_RECEIVE_COUNT = '3'; + worker._run.mockRejectedValue(rateLimited(20 * MINUTE)); + + await run( + sqsRecord({ attributes: { ApproximateReceiveCount: '3' } }) + ); + + expect(worker.recordRateLimitWait.mock.calls[0][2]).toMatchObject({ + status: 'EXHAUSTED', + mechanism: 'visibility', + deferralId: 'msg-1', + }); + }); + + it('clears the wait after a deferred message succeeds, naming the message that set it', async () => { + worker._run.mockResolvedValue(undefined); + + await run( + sqsRecord({ + messageId: 'msg-2', + body: JSON.stringify(deferredBody(1)), + }) + ); + + expect(worker.clearRateLimitWait).toHaveBeenCalledWith( + deferredBody(1), + { deferralId: 'msg-1' } + ); + }); + + it('clears the wait after a redelivered message succeeds, naming that message', async () => { + worker._run.mockResolvedValue(undefined); + + await run( + sqsRecord({ attributes: { ApproximateReceiveCount: '2' } }) + ); + + expect(worker.clearRateLimitWait).toHaveBeenCalledWith(BODY, { + deferralId: 'msg-1', + }); + }); + + it('does not clear anything after a first delivery', async () => { + worker._run.mockResolvedValue(undefined); + + await run(); + + expect(worker.clearRateLimitWait).not.toHaveBeenCalled(); + }); + + it('logs a failing hook and still reports the message handled', async () => { + worker.recordRateLimitWait.mockRejectedValue( + new Error('database is down') + ); + worker._run.mockRejectedValue(rateLimited(60 * SECOND)); + + const result = await run(); + + expect(result).toEqual({ batchItemFailures: [] }); + expect(logged('rate_limit_state_failed')).toEqual([ + expect.objectContaining({ + level: 'WARN', + operation: 'record', + error: expect.objectContaining({ + message: 'database is down', + }), + }), + ]); + }); + + it('logs a failing clear and still counts the message a success', async () => { + worker.clearRateLimitWait.mockRejectedValue( + new Error('database is down') + ); + worker._run.mockResolvedValue(undefined); + + const result = await run( + sqsRecord({ attributes: { ApproximateReceiveCount: '2' } }) + ); + + expect(result).toEqual({ batchItemFailures: [] }); + expect(logged('rate_limit_state_failed')).toEqual([ + expect.objectContaining({ operation: 'clear' }), + ]); + }); + }); + + describe('the SQS client', () => { + it('is built with the offline and endpoint options', () => { + process.env.AWS_ENDPOINT = 'http://localhost:4566'; + process.env.IS_OFFLINE = 'true'; + const constructed = []; + + jest.isolateModules(() => { + jest.doMock('@aws-sdk/client-sqs', () => ({ + ...jest.requireActual('@aws-sdk/client-sqs'), + SQSClient: jest.fn((config) => { + constructed.push(config); + return { send: jest.fn() }; + }), + })); + require('./Worker'); + }); + + expect(constructed.length).toBeGreaterThan(0); + for (const config of constructed) { + expect(config).toMatchObject({ + endpoint: 'http://localhost:4566', + region: 'us-east-1', + }); + } + }); + }); +}); diff --git a/packages/core/core/Worker.js b/packages/core/core/Worker.js index 27c20e864..11f962a7e 100644 --- a/packages/core/core/Worker.js +++ b/packages/core/core/Worker.js @@ -1,14 +1,38 @@ -const { SQSClient, GetQueueUrlCommand, SendMessageCommand } = require('@aws-sdk/client-sqs'); +const { + SQSClient, + GetQueueUrlCommand, + SendMessageCommand, + ChangeMessageVisibilityCommand, +} = require('@aws-sdk/client-sqs'); const _ = require('lodash'); const { RequiredPropertyError } = require('../errors'); const { get } = require('../assertions'); const { readQueueDelivery } = require('../queues/queue-delivery'); +const { awsConfigOptions } = require('../queues/queuer-util'); +const { + MAX_DEFERRED_VISIBILITY_SECONDS, + MAX_DELAY_SECONDS, + deferredMs, + isDeferralCapped, + nextDeferral, + readDeferral, + readDeferralLimits, + withDeferral, +} = require('../queues/queue-deferral'); const { runMessageScope } = require('./invocation-scope'); const { getLogger } = require('../logs'); -const sqs = new SQSClient({ region: process.env.AWS_REGION }); +const sqs = new SQSClient({ + region: process.env.AWS_REGION, + ...awsConfigOptions(), +}); const log = getLogger('frigg.worker'); +const validDate = (value) => + value instanceof Date && !Number.isNaN(value.getTime()) ? value : null; +const queueNameOf = (arn) => + typeof arn === 'string' && arn ? arn.split(':').pop() : undefined; + class Worker { async getQueueURL(params) { // Passing params in because there will be multiple QueueNames @@ -33,12 +57,31 @@ class Worker { // messageId, receiveCount and the event come from the scope. log.debug('Record started', { eventName: 'frigg.worker.record_started' }); + const delivery = readQueueDelivery(record); + let runParams; try { - const runParams = JSON.parse(record.body); + runParams = JSON.parse(record.body); this._validateParams(runParams); - await this._run(runParams, context, readQueueDelivery(record)); + await this._run(runParams, context, delivery); + await this._clearDeferredState(record, runParams, delivery); log.debug('Record succeeded', { eventName: 'frigg.worker.record_succeeded' }); } catch (error) { + let deferralResult; + if (error.isRateLimited && runParams) { + deferralResult = await this.defer( + record, + runParams, + error, + delivery + ); + if (deferralResult.outcome === 'acked') return; + if (deferralResult.outcome === 'failed') { + batchItemFailures.push({ + itemIdentifier: record.messageId, + }); + return; + } + } if (error.isHaltError) { // HaltError means "discard this message, don't retry". // Treat as success so SQS deletes it from the queue. @@ -51,11 +94,30 @@ class Worker { }); return; } - // The message goes back to SQS, so WARN (ADR-048 §4). - log.warn('Record failed, returned for retry', { - eventName: 'frigg.worker.record_failed', - error, - }); + const skippedDeferral = deferralResult && { + deferral: { + skipped: deferralResult.skipped, + ...(deferralResult.cause && { + cause: deferralResult.cause, + }), + }, + }; + if (skippedDeferral && delivery.isLastAttempt) { + this._logLost({ + retryAt: validDate(error.retryAt)?.toISOString(), + reason: error.reason, + module: error.module, + ...skippedDeferral, + error, + }); + } else { + // The message goes back to SQS, so WARN (ADR-048 §4). + log.warn('Record failed, returned for retry', { + eventName: 'frigg.worker.record_failed', + error, + ...skippedDeferral, + }); + } batchItemFailures.push({ itemIdentifier: record.messageId }); } }); @@ -75,6 +137,251 @@ class Worker { // parameters } + async defer(record, body, error, delivery) { + const retryAt = validDate(error.retryAt); + if (!retryAt) return { outcome: 'skipped', skipped: 'no_retry_at' }; + const queueName = queueNameOf(record.eventSourceARN); + if (!queueName || queueName.endsWith('.fifo')) { + return { outcome: 'skipped', skipped: 'no_event_source' }; + } + + const now = Date.now(); + const counters = nextDeferral(body, now, record.messageId); + const deferral = { + record, + queueName, + body, + deferredBody: withDeferral(body, counters), + error, + delivery, + retryAt, + waitMs: Math.max(0, retryAt.getTime() - now), + counters, + }; + + const limits = readDeferralLimits(); + if (isDeferralCapped({ ...counters, retryAt }, limits)) { + await this._recordWait(deferral, 'EXHAUSTED', 'none'); + this._logUnacked( + deferral, + 'record_deferral_capped', + 'Rate-limit deferral capped', + { + maxDeferrals: limits.maxDeferrals, + maxDeferredMs: limits.maxDeferredMs, + deferredMs: deferredMs({ ...counters, retryAt }), + } + ); + return { outcome: 'failed' }; + } + + if (deferral.waitMs <= MAX_DELAY_SECONDS * 1000) + return this._deferWithDelay(deferral); + const scheduled = await this._deferWithSchedule(deferral); + if (scheduled?.outcome) return scheduled; + return this._deferWithVisibility(deferral, scheduled?.scheduleError); + } + + async _deferWithDelay(deferral) { + const delaySeconds = Math.ceil(deferral.waitMs / 1000); + // Write the wait before the send: with a short delay the new message + // can run, and clear the wait, before a later write lands. + await this._recordWait(deferral, 'WAITING', 'delay'); + try { + const queueUrl = await this._queueUrl(deferral.queueName); + await this.send( + { ...deferral.deferredBody, QueueUrl: queueUrl }, + delaySeconds + ); + } catch (cause) { + await this._clearWait(deferral.body, deferral.counters.deferralId); + return { outcome: 'skipped', skipped: 'send_failed', cause }; + } + return this._acknowledgeDeferral(deferral, 'delay', { delaySeconds }); + } + + async _deferWithSchedule(deferral) { + const { record, retryAt, deferredBody } = deferral; + const scheduleName = `frigg-defer-${record.messageId}`; + try { + const scheduler = this.getSchedulerService(); + if (!scheduler) return null; + await scheduler.scheduleOneTime({ + scheduleName, + scheduleAt: retryAt, + queueResourceId: record.eventSourceARN, + payload: deferredBody, + }); + } catch (cause) { + if (cause?.name !== 'ConflictException') + return { scheduleError: cause }; + } + await this._recordWait(deferral, 'WAITING', 'schedule'); + return this._acknowledgeDeferral(deferral, 'schedule', { + scheduleName, + }); + } + + async _deferWithVisibility(deferral, scheduleError) { + const { record, queueName, delivery, waitMs } = deferral; + const visibilityTimeout = Math.min( + Math.ceil(waitMs / 1000), + MAX_DEFERRED_VISIBILITY_SECONDS + ); + try { + const queueUrl = await this._queueUrl(queueName); + await sqs.send( + new ChangeMessageVisibilityCommand({ + QueueUrl: queueUrl, + ReceiptHandle: record.receiptHandle, + VisibilityTimeout: visibilityTimeout, + }) + ); + } catch (cause) { + return { outcome: 'skipped', skipped: 'visibility_failed', cause }; + } + await this._recordWait( + deferral, + delivery.isLastAttempt ? 'EXHAUSTED' : 'WAITING', + 'visibility' + ); + this._logUnacked( + deferral, + 'record_visibility_extended', + 'Record visibility extended', + { + mechanism: 'visibility', + visibilityTimeout, + ...(scheduleError && { scheduleError }), + } + ); + return { outcome: 'failed' }; + } + + async _acknowledgeDeferral(deferral, mechanism, fields) { + log.warn('Record deferred', { + eventName: 'frigg.worker.record_deferred', + mechanism, + ...fields, + ...this._deferralLogFields(deferral), + error: deferral.error, + }); + return { outcome: 'acked' }; + } + + getSchedulerService() { + if (this._schedulerService !== undefined) return this._schedulerService; + if ( + !process.env.SCHEDULER_ROLE_ARN || + process.env.SCHEDULER_PROVIDER === 'mock' + ) { + this._schedulerService = null; + return null; + } + const { + createSchedulerService, + SCHEDULER_PROVIDERS, + } = require('../infrastructure/scheduler/scheduler-service-factory'); + this._schedulerService = createSchedulerService({ + provider: SCHEDULER_PROVIDERS.EVENTBRIDGE, + }); + return this._schedulerService; + } + + recordRateLimitWait() { + return Promise.resolve(); + } + + clearRateLimitWait() { + return Promise.resolve(); + } + + _queueUrl(queueName) { + this._queueUrls ??= new Map(); + if (!this._queueUrls.has(queueName)) { + const lookup = this.getQueueURL({ QueueName: queueName }).catch( + (error) => { + this._queueUrls.delete(queueName); + throw error; + } + ); + this._queueUrls.set(queueName, lookup); + } + return this._queueUrls.get(queueName); + } + + async _recordWait({ body, error, counters, retryAt }, status, mechanism) { + await this._runStateHook('record', () => + this.recordRateLimitWait(body, error, { + status, + mechanism, + deferrals: counters.deferrals, + deferralId: counters.deferralId, + retryAt, + }) + ); + } + + async _clearDeferredState(record, body, delivery) { + const redelivered = delivery.receiveCount > 1; + const { deferrals, deferralId } = readDeferral(body); + if (deferrals === 0 && !redelivered) return; + await this._clearWait(body, deferralId ?? record.messageId); + } + + async _clearWait(body, deferralId) { + await this._runStateHook('clear', () => + this.clearRateLimitWait(body, { deferralId }) + ); + } + + async _runStateHook(operation, hook) { + try { + await hook(); + } catch (error) { + log.warn('Rate-limit run state not written', { + eventName: 'frigg.worker.rate_limit_state_failed', + operation, + error, + }); + } + } + + _deferralLogFields({ error, retryAt, waitMs, counters }) { + return { + waitMs, + retryAt: retryAt.toISOString(), + deferrals: counters.deferrals, + firstDeferredAt: counters.firstDeferredAt, + reason: error.reason, + module: error.module, + statusCode: error.statusCode, + }; + } + + _logUnacked(deferral, eventName, message, fields) { + const logFields = { + ...fields, + ...this._deferralLogFields(deferral), + error: deferral.error, + }; + if (deferral.delivery.isLastAttempt) { + this._logLost(logFields); + return; + } + log.warn(message, { + eventName: `frigg.worker.${eventName}`, + ...logFields, + }); + } + + _logLost(fields) { + log.error( + 'Rate-limited record lost: the message goes to the dead-letter queue next', + { eventName: 'frigg.worker.record_lost_rate_limited', ...fields } + ); + } + // returns the message id async send(params, delay = 0) { this._validateParams(params); diff --git a/packages/core/handlers/backend-utils.js b/packages/core/handlers/backend-utils.js index 1ee6fee44..3f678a6d7 100644 --- a/packages/core/handlers/backend-utils.js +++ b/packages/core/handlers/backend-utils.js @@ -21,6 +21,12 @@ const { getModulesDefinitionFromIntegrationClasses, } = require('../integrations/utils/map-integration-dto'); const { processNotFound } = require('../integrations/use-cases/process-errors'); +const { + RecordRateLimitWait, +} = require('../integrations/use-cases/record-rate-limit-wait'); +const { + ClearRateLimitWait, +} = require('../integrations/use-cases/clear-rate-limit-wait'); const loadRouterFromObject = (IntegrationClass, routerObject) => { const router = Router(); @@ -165,6 +171,28 @@ const createQueueWorker = (integrationClass) => { const integrationName = integrationClass.Definition.name; class QueueWorker extends Worker { + async recordRateLimitWait(body, error, state) { + const processId = body.data?.processId; + if (!processId) return; + const recordRateLimitWait = new RecordRateLimitWait({ + processRepository: createProcessRepository(), + }); + await recordRateLimitWait.execute(processId, { + ...state, + reason: error.reason, + module: error.module, + }); + } + + async clearRateLimitWait(body, { deferralId } = {}) { + const processId = body?.data?.processId; + if (!processId) return; + const clearRateLimitWait = new ClearRateLimitWait({ + processRepository: createProcessRepository(), + }); + await clearRateLimitWait.execute(processId, { deferralId }); + } + async _run(params, context, delivery) { const logCtx = { integration: integrationName, @@ -285,7 +313,8 @@ const createQueueWorker = (integrationClass) => { status >= 400 && status < 500 && status !== 408 && - status !== 429 + status !== 429 && + !error.isRateLimited ) { error.isHaltError = true; console.warn( diff --git a/packages/core/handlers/workers/integration-defined-workers.test.js b/packages/core/handlers/workers/integration-defined-workers.test.js index 940fa2879..e915dfd28 100644 --- a/packages/core/handlers/workers/integration-defined-workers.test.js +++ b/packages/core/handlers/workers/integration-defined-workers.test.js @@ -7,6 +7,7 @@ jest.mock('../../database/config', () => ({ const { createQueueWorker } = require('../backend-utils'); const { IntegrationBase } = require('../../integrations/integration-base'); const { IntegrationEventDispatcher } = require('../integration-event-dispatcher'); +const { RateLimitError } = require('../../errors'); class TestWebhookIntegration extends IntegrationBase { static Definition = { @@ -261,6 +262,42 @@ describe('Webhook Queue Worker', () => { expect(result.batchItemFailures).toEqual([{ itemIdentifier: 'msg-1' }]); }); + it('should NOT mark a 403 with isRateLimited as isHaltError (a provider limit outside 429)', async () => { + const error = new RateLimitError({ + resource: 'https://api.example.com/v1/items', + response: { status: 403, bodyUsed: true }, + hint: { waitMs: 60_000, reason: 'daily', source: 'body' }, + }); + + const FailingIntegration = class extends TestWebhookIntegration { + async onWebhook() { + throw error; + } + }; + + const QueueWorker = createQueueWorker(FailingIntegration); + const worker = new QueueWorker(); + + const sqsEvent = { + Records: [ + { + messageId: 'msg-1', + body: JSON.stringify({ + event: 'ON_WEBHOOK', + data: { body: {} }, + }), + }, + ], + }; + + const result = await worker.run(sqsEvent, {}); + expect(result.batchItemFailures).toEqual([ + { itemIdentifier: 'msg-1' }, + ]); + expect(error.statusCode).toBe(403); + expect(error.isHaltError).toBeUndefined(); + }); + it('should NOT mark 500 as isHaltError (server may recover)', async () => { const error = new Error('Internal Server Error'); error.statusCode = 500; @@ -299,6 +336,210 @@ describe('Webhook Queue Worker', () => { }); }); + describe('Rate-limit run state', () => { + const NOW = Date.parse('2026-09-28T12:00:00.000Z'); + const MINUTE = 60_000; + const at = (ms) => new Date(NOW + ms); + const stored = (retryAt) => ({ + status: 'WAITING', + mechanism: 'delay', + deferrals: 1, + retryAt: retryAt.toISOString(), + reason: 'burst', + module: 'hubspot', + }); + const next = (retryAt) => ({ + status: 'WAITING', + mechanism: 'delay', + deferrals: 1, + retryAt, + }); + const rateLimitError = Object.assign(new Error('limited'), { + isRateLimited: true, + reason: 'burst', + module: 'hubspot', + }); + const message = { data: { processId: 7 } }; + let nowSpy; + + beforeEach(() => { + nowSpy = jest.spyOn(Date, 'now').mockReturnValue(NOW); + }); + + afterEach(() => nowSpy.mockRestore()); + + function buildWorker(rateLimit) { + const applyProcessUpdate = jest.fn().mockResolvedValue({ id: 7 }); + const findById = jest.fn().mockResolvedValue({ + id: 7, + context: rateLimit === undefined ? {} : { rateLimit }, + }); + let QueueWorker; + jest.isolateModules(() => { + jest.doMock( + '../../integrations/repositories/process-repository-factory', + () => ({ + createProcessRepository: () => ({ + applyProcessUpdate, + findById, + }), + }) + ); + jest.doMock( + '../../integrations/repositories/integration-repository-factory', + () => ({ createIntegrationRepository: () => ({}) }) + ); + jest.doMock( + '../../modules/repositories/module-repository-factory', + () => ({ createModuleRepository: () => ({}) }) + ); + jest.doMock( + '../../integrations/use-cases/get-integration-instance', + () => ({ + GetIntegrationInstance: class { + async execute() { + return new TestWebhookIntegration(); + } + }, + }) + ); + QueueWorker = require('../backend-utils').createQueueWorker( + TestWebhookIntegration + ); + }); + return { worker: new QueueWorker(), applyProcessUpdate, findById }; + } + + describe('recording a wait', () => { + it('writes context.rateLimit on the process of a deferred message', async () => { + const { worker, applyProcessUpdate } = buildWorker(); + + await worker.recordRateLimitWait(message, rateLimitError, { + ...next(at(20 * MINUTE)), + deferralId: 'msg-1', + }); + + expect(applyProcessUpdate).toHaveBeenCalledWith(7, { + set: { + 'context.rateLimit': { + status: 'WAITING', + mechanism: 'delay', + retryAt: '2026-09-28T12:20:00.000Z', + reason: 'burst', + module: 'hubspot', + deferrals: 1, + deferralId: 'msg-1', + updatedAt: expect.any(String), + }, + }, + }); + }); + + it('writes nothing for a message with no processId', async () => { + const { worker, applyProcessUpdate, findById } = buildWorker(); + + await worker.recordRateLimitWait( + { data: { integrationId: 3 } }, + rateLimitError, + next(at(MINUTE)) + ); + + expect(findById).not.toHaveBeenCalled(); + expect(applyProcessUpdate).not.toHaveBeenCalled(); + }); + }); + + describe('clearing a wait', () => { + const sqsRecord = (frigg) => ({ + messageId: 'msg-1', + body: JSON.stringify({ + event: 'ON_WEBHOOK', + data: { processId: 7 }, + ...(frigg && { _frigg: frigg }), + }), + attributes: { ApproximateReceiveCount: '1' }, + }); + + it('clears the wait after a deferred message runs', async () => { + const { worker, applyProcessUpdate } = buildWorker( + stored(at(-MINUTE)) + ); + + const result = await worker.run({ + Records: [ + sqsRecord({ + deferrals: 1, + firstDeferredAt: at(-20 * MINUTE).toISOString(), + }), + ], + }); + + expect(result.batchItemFailures).toEqual([]); + expect(applyProcessUpdate).toHaveBeenCalledWith(7, { + set: { 'context.rateLimit': null }, + }); + }); + + it('clears at once the wait that the finished message set, before its retry time', async () => { + const { worker, applyProcessUpdate } = buildWorker({ + ...stored(at(60 * MINUTE)), + deferralId: 'msg-1', + }); + + const result = await worker.run({ + Records: [ + { + ...sqsRecord({ + deferrals: 1, + firstDeferredAt: at(-MINUTE).toISOString(), + deferralId: 'msg-1', + }), + messageId: 'msg-2', + }, + ], + }); + + expect(result.batchItemFailures).toEqual([]); + expect(applyProcessUpdate).toHaveBeenCalledWith(7, { + set: { 'context.rateLimit': null }, + }); + }); + + it('does not clear the wait after a first delivery runs', async () => { + const { worker, applyProcessUpdate } = buildWorker( + stored(at(-MINUTE)) + ); + + const result = await worker.run({ Records: [sqsRecord()] }); + + expect(result.batchItemFailures).toEqual([]); + expect(applyProcessUpdate).not.toHaveBeenCalled(); + }); + + it('clears a wait that is over', async () => { + const { worker, applyProcessUpdate } = buildWorker( + stored(at(-MINUTE)) + ); + + await worker.clearRateLimitWait(message); + + expect(applyProcessUpdate).toHaveBeenCalledWith(7, { + set: { 'context.rateLimit': null }, + }); + }); + + it('writes nothing for a message with no processId', async () => { + const { worker, applyProcessUpdate, findById } = buildWorker(); + + await worker.clearRateLimitWait({ data: {} }); + await worker.clearRateLimitWait({}); + + expect(findById).not.toHaveBeenCalled(); + expect(applyProcessUpdate).not.toHaveBeenCalled(); + }); + }); + }); + describe('Integration Hydration for webhooks with integrationId', () => { it('should attempt to load integration when integrationId present', async () => { // This test verifies the logic path - full integration test diff --git a/packages/core/infrastructure/scheduler/eventbridge-scheduler-adapter.js b/packages/core/infrastructure/scheduler/eventbridge-scheduler-adapter.js index c06c46375..d7ac3b1ea 100644 --- a/packages/core/infrastructure/scheduler/eventbridge-scheduler-adapter.js +++ b/packages/core/infrastructure/scheduler/eventbridge-scheduler-adapter.js @@ -20,12 +20,14 @@ const { } = require('@aws-sdk/client-scheduler'); const { SchedulerServiceInterface } = require('./scheduler-service-interface'); +const { awsConfigOptions } = require('../../queues/queuer-util'); class EventBridgeSchedulerAdapter extends SchedulerServiceInterface { constructor({ region } = {}) { super(); this.client = new SchedulerClient({ region: region || process.env.AWS_REGION || 'us-east-1', + ...awsConfigOptions(), }); this.scheduleGroupName = process.env.SCHEDULE_GROUP_NAME || 'frigg-integration-schedules'; diff --git a/packages/core/infrastructure/scheduler/eventbridge-scheduler-adapter.test.js b/packages/core/infrastructure/scheduler/eventbridge-scheduler-adapter.test.js new file mode 100644 index 000000000..d3b2f85a6 --- /dev/null +++ b/packages/core/infrastructure/scheduler/eventbridge-scheduler-adapter.test.js @@ -0,0 +1,46 @@ +const { + EventBridgeSchedulerAdapter, +} = require('./eventbridge-scheduler-adapter'); + +const ENV_KEYS = ['IS_OFFLINE', 'AWS_ENDPOINT', 'AWS_REGION']; + +describe('EventBridgeSchedulerAdapter', () => { + let savedEnv; + + beforeEach(() => { + savedEnv = {}; + for (const key of ENV_KEYS) { + savedEnv[key] = process.env[key]; + delete process.env[key]; + } + }); + + afterEach(() => { + for (const key of ENV_KEYS) { + if (savedEnv[key] === undefined) delete process.env[key]; + else process.env[key] = savedEnv[key]; + } + }); + + it('uses the offline credentials and the local endpoint, as the queue clients do', async () => { + process.env.IS_OFFLINE = 'true'; + process.env.AWS_ENDPOINT = 'http://localhost:4566'; + + const { client } = new EventBridgeSchedulerAdapter(); + + const endpoint = await client.config.endpoint(); + expect(endpoint).toMatchObject({ hostname: 'localhost', port: 4566 }); + await expect(client.config.credentials()).resolves.toMatchObject({ + accessKeyId: 'test-aws-key', + }); + }); + + it('keeps the AWS endpoint and the region when no local endpoint is set', async () => { + process.env.AWS_REGION = 'eu-west-1'; + + const { client } = new EventBridgeSchedulerAdapter(); + + expect(client.config.isCustomEndpoint).toBe(false); + await expect(client.config.region()).resolves.toBe('eu-west-1'); + }); +}); diff --git a/packages/core/integrations/use-cases/clear-rate-limit-wait.js b/packages/core/integrations/use-cases/clear-rate-limit-wait.js new file mode 100644 index 000000000..b243c574b --- /dev/null +++ b/packages/core/integrations/use-cases/clear-rate-limit-wait.js @@ -0,0 +1,28 @@ +const CLEAR_RATE_LIMIT_TOLERANCE_MS = 5_000; + +const isOwnWait = (current, deferralId) => + Boolean(deferralId) && current.deferralId === deferralId; + +const isOver = (current) => + Date.parse(current.retryAt) - Date.now() <= CLEAR_RATE_LIMIT_TOLERANCE_MS; + +class ClearRateLimitWait { + constructor({ processRepository }) { + if (!processRepository) { + throw new Error('processRepository is required'); + } + this.processRepository = processRepository; + } + + async execute(processId, { deferralId } = {}) { + const process = await this.processRepository.findById(processId); + const current = process?.context?.rateLimit; + if (current?.status !== 'WAITING') return; + if (!isOwnWait(current, deferralId) && !isOver(current)) return; + await this.processRepository.applyProcessUpdate(processId, { + set: { 'context.rateLimit': null }, + }); + } +} + +module.exports = { ClearRateLimitWait }; diff --git a/packages/core/integrations/use-cases/clear-rate-limit-wait.test.js b/packages/core/integrations/use-cases/clear-rate-limit-wait.test.js new file mode 100644 index 000000000..2d9b8ea95 --- /dev/null +++ b/packages/core/integrations/use-cases/clear-rate-limit-wait.test.js @@ -0,0 +1,143 @@ +const { ClearRateLimitWait } = require('./clear-rate-limit-wait'); + +describe('ClearRateLimitWait', () => { + const NOW = Date.parse('2026-09-28T12:00:00.000Z'); + const MINUTE = 60_000; + const processId = 'process-123'; + const at = (ms) => new Date(NOW + ms); + const stored = (retryAt, status = 'WAITING') => ({ + status, + mechanism: 'delay', + deferrals: 1, + retryAt: retryAt.toISOString(), + reason: 'burst', + module: 'hubspot', + }); + let clearRateLimitWait; + let mockProcessRepository; + let nowSpy; + + const storeOnProcess = (rateLimit) => + mockProcessRepository.findById.mockResolvedValue({ + id: processId, + context: { rateLimit }, + }); + + beforeEach(() => { + nowSpy = jest.spyOn(Date, 'now').mockReturnValue(NOW); + mockProcessRepository = { + findById: jest + .fn() + .mockResolvedValue({ id: processId, context: {} }), + applyProcessUpdate: jest.fn().mockResolvedValue({ id: processId }), + }; + clearRateLimitWait = new ClearRateLimitWait({ + processRepository: mockProcessRepository, + }); + }); + + afterEach(() => nowSpy.mockRestore()); + + describe('constructor', () => { + it('should require processRepository', () => { + expect(() => new ClearRateLimitWait({})).toThrow( + 'processRepository is required' + ); + }); + + it('should initialize with processRepository', () => { + expect(clearRateLimitWait.processRepository).toBe( + mockProcessRepository + ); + }); + }); + + describe('execute', () => { + it('clears a wait that is over', async () => { + storeOnProcess(stored(at(-MINUTE))); + + await clearRateLimitWait.execute(processId); + + expect( + mockProcessRepository.applyProcessUpdate + ).toHaveBeenCalledWith(processId, { + set: { 'context.rateLimit': null }, + }); + }); + + it('clears a wait that ends within a few seconds', async () => { + storeOnProcess(stored(at(3_000))); + + await clearRateLimitWait.execute(processId); + + expect( + mockProcessRepository.applyProcessUpdate + ).toHaveBeenCalledTimes(1); + }); + + it('keeps a wait that other messages still wait for', async () => { + storeOnProcess(stored(at(10 * MINUTE))); + + await clearRateLimitWait.execute(processId); + + expect( + mockProcessRepository.applyProcessUpdate + ).not.toHaveBeenCalled(); + }); + + it('clears at once the wait that the finished message set', async () => { + storeOnProcess({ ...stored(at(60 * MINUTE)), deferralId: 'msg-1' }); + + await clearRateLimitWait.execute(processId, { + deferralId: 'msg-1', + }); + + expect( + mockProcessRepository.applyProcessUpdate + ).toHaveBeenCalledWith(processId, { + set: { 'context.rateLimit': null }, + }); + }); + + it('keeps a wait that another message set until its retry time', async () => { + storeOnProcess({ ...stored(at(60 * MINUTE)), deferralId: 'msg-1' }); + + await clearRateLimitWait.execute(processId, { + deferralId: 'msg-2', + }); + + expect( + mockProcessRepository.applyProcessUpdate + ).not.toHaveBeenCalled(); + }); + + it('leaves a cap in place', async () => { + storeOnProcess(stored(at(-MINUTE), 'EXHAUSTED')); + + await clearRateLimitWait.execute(processId); + + expect( + mockProcessRepository.applyProcessUpdate + ).not.toHaveBeenCalled(); + }); + + it('writes nothing when no wait is recorded', async () => { + await clearRateLimitWait.execute(processId); + + expect( + mockProcessRepository.applyProcessUpdate + ).not.toHaveBeenCalled(); + }); + + it('does not fail when the process is gone', async () => { + mockProcessRepository.findById.mockResolvedValue(null); + + await expect( + clearRateLimitWait.execute(processId) + ).resolves.toBeUndefined(); + expect( + mockProcessRepository.applyProcessUpdate + ).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/packages/core/integrations/use-cases/index.js b/packages/core/integrations/use-cases/index.js index d7ce7a7fc..9c33015f7 100644 --- a/packages/core/integrations/use-cases/index.js +++ b/packages/core/integrations/use-cases/index.js @@ -6,6 +6,8 @@ const { CreateProcess } = require('./create-process'); const { UpdateProcessState } = require('./update-process-state'); const { UpdateProcessMetrics } = require('./update-process-metrics'); const { GetProcess } = require('./get-process'); +const { RecordRateLimitWait } = require('./record-rate-limit-wait'); +const { ClearRateLimitWait } = require('./clear-rate-limit-wait'); module.exports = { GetIntegrationsForUser, @@ -16,4 +18,6 @@ module.exports = { UpdateProcessState, UpdateProcessMetrics, GetProcess, -}; \ No newline at end of file + RecordRateLimitWait, + ClearRateLimitWait, +}; \ No newline at end of file diff --git a/packages/core/integrations/use-cases/record-rate-limit-wait.js b/packages/core/integrations/use-cases/record-rate-limit-wait.js new file mode 100644 index 000000000..05b83f4da --- /dev/null +++ b/packages/core/integrations/use-cases/record-rate-limit-wait.js @@ -0,0 +1,36 @@ +const keepsCurrentRateLimit = (current, next) => { + if (!current) return false; + if (next.status === 'EXHAUSTED') return false; + if (current.status === 'EXHAUSTED') return true; + return Date.parse(current.retryAt) > next.retryAt.getTime(); +}; + +class RecordRateLimitWait { + constructor({ processRepository }) { + if (!processRepository) { + throw new Error('processRepository is required'); + } + this.processRepository = processRepository; + } + + async execute(processId, wait) { + const process = await this.processRepository.findById(processId); + if (keepsCurrentRateLimit(process?.context?.rateLimit, wait)) return; + await this.processRepository.applyProcessUpdate(processId, { + set: { + 'context.rateLimit': { + status: wait.status, + mechanism: wait.mechanism, + retryAt: wait.retryAt.toISOString(), + reason: wait.reason, + module: wait.module, + deferrals: wait.deferrals, + deferralId: wait.deferralId, + updatedAt: new Date().toISOString(), + }, + }, + }); + } +} + +module.exports = { RecordRateLimitWait }; diff --git a/packages/core/integrations/use-cases/record-rate-limit-wait.test.js b/packages/core/integrations/use-cases/record-rate-limit-wait.test.js new file mode 100644 index 000000000..43bef1cbe --- /dev/null +++ b/packages/core/integrations/use-cases/record-rate-limit-wait.test.js @@ -0,0 +1,145 @@ +const { RecordRateLimitWait } = require('./record-rate-limit-wait'); + +describe('RecordRateLimitWait', () => { + const NOW = Date.parse('2026-09-28T12:00:00.000Z'); + const MINUTE = 60_000; + const processId = 'process-123'; + const at = (ms) => new Date(NOW + ms); + const stored = (retryAt, status = 'WAITING') => ({ + status, + mechanism: 'delay', + deferrals: 1, + retryAt: retryAt.toISOString(), + reason: 'burst', + module: 'hubspot', + }); + const wait = (retryAt, status = 'WAITING') => ({ + status, + mechanism: 'delay', + deferrals: 1, + retryAt, + reason: 'burst', + module: 'hubspot', + }); + let recordRateLimitWait; + let mockProcessRepository; + + const storeOnProcess = (rateLimit) => + mockProcessRepository.findById.mockResolvedValue({ + id: processId, + context: { rateLimit }, + }); + + beforeEach(() => { + mockProcessRepository = { + findById: jest + .fn() + .mockResolvedValue({ id: processId, context: {} }), + applyProcessUpdate: jest.fn().mockResolvedValue({ id: processId }), + }; + recordRateLimitWait = new RecordRateLimitWait({ + processRepository: mockProcessRepository, + }); + }); + + describe('constructor', () => { + it('should require processRepository', () => { + expect(() => new RecordRateLimitWait({})).toThrow( + 'processRepository is required' + ); + }); + + it('should initialize with processRepository', () => { + expect(recordRateLimitWait.processRepository).toBe( + mockProcessRepository + ); + }); + }); + + describe('execute', () => { + it('writes the wait as context.rateLimit on the process', async () => { + await recordRateLimitWait.execute(processId, wait(at(20 * MINUTE))); + + expect( + mockProcessRepository.applyProcessUpdate + ).toHaveBeenCalledWith(processId, { + set: { + 'context.rateLimit': { + status: 'WAITING', + mechanism: 'delay', + retryAt: '2026-09-28T12:20:00.000Z', + reason: 'burst', + module: 'hubspot', + deferrals: 1, + updatedAt: expect.any(String), + }, + }, + }); + }); + + it('keeps the id of the message that set the wait', async () => { + await recordRateLimitWait.execute(processId, { + ...wait(at(20 * MINUTE)), + deferralId: 'msg-1', + }); + + const [, update] = + mockProcessRepository.applyProcessUpdate.mock.calls[0]; + expect(update.set['context.rateLimit']).toMatchObject({ + deferralId: 'msg-1', + }); + }); + + it('replaces an earlier wait', async () => { + storeOnProcess(stored(at(5 * MINUTE))); + + await recordRateLimitWait.execute(processId, wait(at(20 * MINUTE))); + + expect( + mockProcessRepository.applyProcessUpdate + ).toHaveBeenCalledTimes(1); + }); + + it('keeps a later wait that another message recorded', async () => { + storeOnProcess(stored(at(40 * MINUTE))); + + await recordRateLimitWait.execute(processId, wait(at(20 * MINUTE))); + + expect( + mockProcessRepository.applyProcessUpdate + ).not.toHaveBeenCalled(); + }); + + it('shows a cap even when a later wait is recorded', async () => { + storeOnProcess(stored(at(40 * MINUTE))); + + await recordRateLimitWait.execute( + processId, + wait(at(20 * MINUTE), 'EXHAUSTED') + ); + + expect( + mockProcessRepository.applyProcessUpdate + ).toHaveBeenCalledTimes(1); + }); + + it('never hides a cap behind a later wait', async () => { + storeOnProcess(stored(at(5 * MINUTE), 'EXHAUSTED')); + + await recordRateLimitWait.execute(processId, wait(at(20 * MINUTE))); + + expect( + mockProcessRepository.applyProcessUpdate + ).not.toHaveBeenCalled(); + }); + + it('does not fail when the process is gone', async () => { + mockProcessRepository.findById.mockResolvedValue(null); + mockProcessRepository.applyProcessUpdate.mockResolvedValue(null); + + await expect( + recordRateLimitWait.execute(processId, wait(at(MINUTE))) + ).resolves.toBeUndefined(); + }); + }); +}); diff --git a/packages/core/modules/requester/oauth-2.js b/packages/core/modules/requester/oauth-2.js index 2bafb9d84..352df70e4 100644 --- a/packages/core/modules/requester/oauth-2.js +++ b/packages/core/modules/requester/oauth-2.js @@ -398,6 +398,13 @@ class OAuth2Requester extends Requester { ); transportError.statusCode = status; transportError.isTokenRefreshTransportFailure = true; + if (error?.isRateLimited) { + transportError.isRateLimited = true; + transportError.retryAt = error.retryAt; + transportError.waitMs = error.waitMs; + transportError.reason = error.reason; + transportError.module = error.module; + } return transportError; } diff --git a/packages/core/modules/requester/oauth-2.test.js b/packages/core/modules/requester/oauth-2.test.js index 9e55a0059..799cc7216 100644 --- a/packages/core/modules/requester/oauth-2.test.js +++ b/packages/core/modules/requester/oauth-2.test.js @@ -1,4 +1,5 @@ const { OAuth2Requester } = require('./oauth-2'); +const { RateLimitError } = require('../../errors'); describe('OAuth2Requester', () => { describe('constructor', () => { @@ -175,6 +176,75 @@ describe('OAuth2Requester', () => { ); }); + it('keeps the rate-limit fields on the failure of a throttled token endpoint', async () => { + const requester = new OAuth2Requester({ + grant_type: 'authorization_code', + refresh_token: 'test-refresh-token', + credentialReloadBackoffMs: [], + }); + const retryAt = new Date(Date.now() + 3_600_000); + requester.refreshAccessToken = jest.fn().mockRejectedValue( + new RateLimitError({ + resource: 'https://auth.example.com/token', + response: { status: 429, bodyUsed: true }, + hint: { retryAt, reason: 'burst', source: 'header' }, + }) + ); + requester.notify = jest.fn(); + + const error = await requester.refreshAuth().catch((e) => e); + + expect(error).toMatchObject({ + isRateLimited: true, + statusCode: 429, + reason: 'burst', + }); + expect(error.retryAt).toEqual(retryAt); + expect(error.waitMs).toBeGreaterThan(0); + expect(error.message).not.toContain('auth.example.com'); + expect(requester.notify).not.toHaveBeenCalledWith( + requester.DLGT_INVALID_AUTH, + expect.anything() + ); + }); + + it('keeps the module name on the failure of a throttled token endpoint', async () => { + const requester = new OAuth2Requester({ + grant_type: 'authorization_code', + refresh_token: 'test-refresh-token', + credentialReloadBackoffMs: [], + }); + requester.refreshAccessToken = jest.fn().mockRejectedValue( + new RateLimitError({ + resource: 'https://auth.example.com/token', + response: { status: 429, bodyUsed: true }, + waitMs: 3_600_000, + module: 'crm', + }) + ); + + const error = await requester.refreshAuth().catch((e) => e); + + expect(error).toMatchObject({ isRateLimited: true, module: 'crm' }); + }); + + it('does not flag an ordinary transport failure as rate limited', async () => { + const requester = new OAuth2Requester({ + grant_type: 'authorization_code', + refresh_token: 'test-refresh-token', + credentialReloadBackoffMs: [], + }); + requester.refreshAccessToken = jest + .fn() + .mockRejectedValue(new Error('socket hang up')); + requester.notify = jest.fn(); + + const error = await requester.refreshAuth().catch((e) => e); + + expect(error.isRateLimited).toBeUndefined(); + expect(error.retryAt).toBeUndefined(); + }); + it('should return false and notify DLGT_INVALID_AUTH on error during client_credentials refresh', async () => { const requester = new OAuth2Requester({ grant_type: 'client_credentials', diff --git a/packages/core/package.json b/packages/core/package.json index dbbed0af0..bd599c6a0 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -7,6 +7,7 @@ "@aws-sdk/client-kms": "^3.588.0", "@aws-sdk/client-lambda": "^3.714.0", "@aws-sdk/client-s3": "^3.588.0", + "@aws-sdk/client-scheduler": "^3.588.0", "@aws-sdk/client-sqs": "^3.588.0", "@aws-sdk/client-ssm": "^3.588.0", "@aws-sdk/s3-request-presigner": "^3.588.0", @@ -53,6 +54,7 @@ "@prisma/client": "^6.19.3", "@types/lodash": "4.17.15", "@typescript-eslint/eslint-plugin": "^8.0.0", + "aws-sdk-client-mock": "^4.1.0", "chai": "^4.3.6", "eslint": "^8.22.0", "eslint-plugin-import": "^2.29.1", diff --git a/packages/core/queues/queue-deferral.js b/packages/core/queues/queue-deferral.js new file mode 100644 index 000000000..a5d0c35af --- /dev/null +++ b/packages/core/queues/queue-deferral.js @@ -0,0 +1,74 @@ +const { toPositiveInteger } = require('./queue-delivery'); + +const MAX_DEFERRALS = 30; +const MAX_DEFERRED_MS = 26 * 60 * 60 * 1000; +const MAX_DELAY_SECONDS = 900; +const MAX_VISIBILITY_TIMEOUT_SECONDS = 43_200; +const MAX_LAMBDA_RUN_SECONDS = 900; +const MAX_BATCHING_WINDOW_SECONDS = 300; +const MAX_DEFERRED_VISIBILITY_SECONDS = + MAX_VISIBILITY_TIMEOUT_SECONDS - + MAX_LAMBDA_RUN_SECONDS - + MAX_BATCHING_WINDOW_SECONDS; +const MAX_DEFERRALS_ENV = 'FRIGG_QUEUE_MAX_DEFERRALS'; +const MAX_DEFERRED_MS_ENV = 'FRIGG_QUEUE_MAX_DEFERRED_MS'; + +const readDeferralLimits = (env = process.env) => ({ + maxDeferrals: toPositiveInteger(env[MAX_DEFERRALS_ENV]) ?? MAX_DEFERRALS, + maxDeferredMs: + toPositiveInteger(env[MAX_DEFERRED_MS_ENV]) ?? MAX_DEFERRED_MS, +}); + +const readDeferral = (body) => { + const frigg = body?._frigg; + const deferrals = + Number.isInteger(frigg?.deferrals) && frigg.deferrals > 0 + ? frigg.deferrals + : 0; + const firstDeferredAt = + typeof frigg?.firstDeferredAt === 'string' && + !Number.isNaN(Date.parse(frigg.firstDeferredAt)) + ? frigg.firstDeferredAt + : undefined; + const deferralId = + typeof frigg?.deferralId === 'string' && frigg.deferralId !== '' + ? frigg.deferralId + : undefined; + return { deferrals, firstDeferredAt, deferralId }; +}; + +const nextDeferral = (body, now = Date.now(), messageId) => { + const current = readDeferral(body); + return { + deferrals: current.deferrals + 1, + firstDeferredAt: current.firstDeferredAt ?? new Date(now).toISOString(), + deferralId: current.deferralId ?? messageId, + }; +}; + +const withDeferral = (body, deferral) => ({ + ...body, + _frigg: { ...body._frigg, ...deferral }, +}); + +const deferredMs = ({ firstDeferredAt, retryAt }) => + retryAt.getTime() - Date.parse(firstDeferredAt); + +const isDeferralCapped = (deferral, { maxDeferrals, maxDeferredMs }) => + deferral.deferrals > maxDeferrals || deferredMs(deferral) > maxDeferredMs; + +module.exports = { + MAX_DEFERRALS, + MAX_DEFERRALS_ENV, + MAX_DEFERRED_MS, + MAX_DEFERRED_MS_ENV, + MAX_DEFERRED_VISIBILITY_SECONDS, + MAX_DELAY_SECONDS, + MAX_VISIBILITY_TIMEOUT_SECONDS, + deferredMs, + isDeferralCapped, + nextDeferral, + readDeferral, + readDeferralLimits, + withDeferral, +}; diff --git a/packages/core/queues/queue-deferral.test.js b/packages/core/queues/queue-deferral.test.js new file mode 100644 index 000000000..643d7bf22 --- /dev/null +++ b/packages/core/queues/queue-deferral.test.js @@ -0,0 +1,192 @@ +const { + MAX_DEFERRALS, + MAX_DEFERRALS_ENV, + MAX_DEFERRED_MS, + MAX_DEFERRED_MS_ENV, + MAX_DELAY_SECONDS, + MAX_VISIBILITY_TIMEOUT_SECONDS, + isDeferralCapped, + nextDeferral, + readDeferral, + readDeferralLimits, + withDeferral, +} = require('./queue-deferral'); + +const NOW = Date.parse('2026-09-28T12:00:00.000Z'); +const HOUR = 60 * 60 * 1000; + +describe('queue-deferral', () => { + it('exposes the SQS limits and the default caps', () => { + expect(MAX_DELAY_SECONDS).toBe(900); + expect(MAX_VISIBILITY_TIMEOUT_SECONDS).toBe(43_200); + expect(MAX_DEFERRALS).toBe(30); + expect(MAX_DEFERRED_MS).toBe(26 * HOUR); + expect(MAX_DEFERRALS_ENV).toBe('FRIGG_QUEUE_MAX_DEFERRALS'); + expect(MAX_DEFERRED_MS_ENV).toBe('FRIGG_QUEUE_MAX_DEFERRED_MS'); + }); + + describe('readDeferralLimits', () => { + it('returns the defaults when the environment sets nothing', () => { + expect(readDeferralLimits({})).toEqual({ + maxDeferrals: 30, + maxDeferredMs: 26 * HOUR, + }); + }); + + it('reads both limits from the environment', () => { + expect( + readDeferralLimits({ + [MAX_DEFERRALS_ENV]: '3', + [MAX_DEFERRED_MS_ENV]: '7200000', + }) + ).toEqual({ maxDeferrals: 3, maxDeferredMs: 7_200_000 }); + }); + + it.each(['0', '-1', '1.5', 'many', ''])( + 'ignores the value %j', + (value) => { + expect( + readDeferralLimits({ + [MAX_DEFERRALS_ENV]: value, + [MAX_DEFERRED_MS_ENV]: value, + }) + ).toEqual({ maxDeferrals: 30, maxDeferredMs: 26 * HOUR }); + } + ); + + it('reads process.env when no environment is given', () => { + const previous = process.env[MAX_DEFERRALS_ENV]; + process.env[MAX_DEFERRALS_ENV] = '4'; + try { + expect(readDeferralLimits().maxDeferrals).toBe(4); + } finally { + if (previous === undefined) + delete process.env[MAX_DEFERRALS_ENV]; + else process.env[MAX_DEFERRALS_ENV] = previous; + } + }); + }); + + describe('readDeferral', () => { + it('reads zero deferrals from a body with no _frigg', () => { + expect(readDeferral({ event: 'X' })).toEqual({ + deferrals: 0, + firstDeferredAt: undefined, + }); + expect(readDeferral(undefined)).toEqual({ + deferrals: 0, + firstDeferredAt: undefined, + }); + }); + + it('reads the counters of a deferred body', () => { + const firstDeferredAt = new Date(NOW).toISOString(); + expect( + readDeferral({ _frigg: { deferrals: 2, firstDeferredAt } }) + ).toEqual({ deferrals: 2, firstDeferredAt }); + }); + + it.each([ + [{ deferrals: -1 }], + [{ deferrals: 1.5 }], + [{ deferrals: 'two' }], + [{ firstDeferredAt: 'yesterday' }], + [{ firstDeferredAt: 5 }], + ])('ignores an invalid counter %j', (frigg) => { + expect(readDeferral({ _frigg: frigg })).toEqual({ + deferrals: 0, + firstDeferredAt: undefined, + }); + }); + }); + + describe('nextDeferral', () => { + it('starts at 1 and stamps the first deferral time', () => { + expect(nextDeferral({ event: 'X' }, NOW)).toEqual({ + deferrals: 1, + firstDeferredAt: new Date(NOW).toISOString(), + }); + }); + + it('adds one and keeps the first deferral time', () => { + const firstDeferredAt = new Date(NOW - HOUR).toISOString(); + expect( + nextDeferral({ _frigg: { deferrals: 2, firstDeferredAt } }, NOW) + ).toEqual({ deferrals: 3, firstDeferredAt }); + }); + }); + + describe('withDeferral', () => { + it('puts the counters in _frigg and keeps the rest of the body', () => { + const body = { + event: 'X', + data: { processId: 4 }, + _frigg: { other: true }, + }; + const deferral = { + deferrals: 1, + firstDeferredAt: new Date(NOW).toISOString(), + }; + expect(withDeferral(body, deferral)).toEqual({ + event: 'X', + data: { processId: 4 }, + _frigg: { other: true, ...deferral }, + }); + expect(body._frigg).toEqual({ other: true }); + }); + }); + + describe('isDeferralCapped', () => { + const limits = { maxDeferrals: 3, maxDeferredMs: 2 * HOUR }; + const first = new Date(NOW).toISOString(); + + it('does not cap a deferral inside both limits', () => { + expect( + isDeferralCapped( + { + deferrals: 3, + firstDeferredAt: first, + retryAt: new Date(NOW + HOUR), + }, + limits + ) + ).toBe(false); + }); + + it('caps a deferral past the count', () => { + expect( + isDeferralCapped( + { + deferrals: 4, + firstDeferredAt: first, + retryAt: new Date(NOW + 1_000), + }, + limits + ) + ).toBe(true); + }); + + it('caps a deferral whose retry time is past the total time, counted from the first deferral', () => { + expect( + isDeferralCapped( + { + deferrals: 2, + firstDeferredAt: first, + retryAt: new Date(NOW + 2 * HOUR + 1), + }, + limits + ) + ).toBe(true); + expect( + isDeferralCapped( + { + deferrals: 2, + firstDeferredAt: first, + retryAt: new Date(NOW + 2 * HOUR), + }, + limits + ) + ).toBe(false); + }); + }); +}); diff --git a/packages/core/queues/queue-delivery.js b/packages/core/queues/queue-delivery.js index 8b5bebb9f..5e2e02299 100644 --- a/packages/core/queues/queue-delivery.js +++ b/packages/core/queues/queue-delivery.js @@ -38,4 +38,5 @@ module.exports = { INTEGRATION_QUEUE_MAX_RECEIVE_COUNT, QUEUE_MAX_RECEIVE_COUNT_ENV, readQueueDelivery, + toPositiveInteger, }; diff --git a/packages/core/queues/queuer-util.js b/packages/core/queues/queuer-util.js index 0c9f502ca..d2f19ad6e 100644 --- a/packages/core/queues/queuer-util.js +++ b/packages/core/queues/queuer-util.js @@ -126,4 +126,4 @@ const QueuerUtil = { }, }; -module.exports = { QueuerUtil }; +module.exports = { QueuerUtil, awsConfigOptions }; diff --git a/packages/core/types/core/index.d.ts b/packages/core/types/core/index.d.ts index 084ac9723..565e531f1 100644 --- a/packages/core/types/core/index.d.ts +++ b/packages/core/types/core/index.d.ts @@ -47,8 +47,27 @@ declare module "@friggframework/core" { send(params: object & { QueueUrl: any }, delay?: number): Promise; sendAsyncSQSMessage(params: SendSQSMessageParams): Promise; + + recordRateLimitWait( + body: object, + error: Error, + state: RateLimitWaitState + ): Promise; + + clearRateLimitWait( + body: object, + options?: { deferralId?: string } + ): Promise; } + export type RateLimitWaitState = { + status: "WAITING" | "EXHAUSTED"; + mechanism: "delay" | "schedule" | "visibility" | "none"; + deferrals: number; + deferralId?: string; + retryAt: Date; + }; + interface IWorker { getQueueURL(params: GetQueueURLParams): Promise; run(params: { Records: any }, context?: object): Promise; diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js index cc0f1c71b..57099a539 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js @@ -10,6 +10,8 @@ const { } = require('@friggframework/core/queues/queue-delivery'); const { IntegrationBuilder } = require('./integration-builder'); const { ValidationResult } = require('../shared/base-builder'); +const { BuilderOrchestrator } = require('../shared/builder-orchestrator'); +const { SchedulerBuilder } = require('../scheduler/scheduler-builder'); describe('IntegrationBuilder', () => { let integrationBuilder; @@ -259,6 +261,37 @@ describe('IntegrationBuilder', () => { ).toBeUndefined(); }); + it('adds no resource, function or env var when no module declares rateLimit', async () => { + class Api {} + const { merged } = await new BuilderOrchestrator([ + new IntegrationBuilder(), + new SchedulerBuilder(), + ]).buildAll({ + integrations: [ + { + Definition: { + name: 'test', + modules: { crm: { definition: { API: Api } } }, + }, + }, + ], + }); + const before = await integrationBuilder.build( + { integrations: [{ Definition: { name: 'test' } }] }, + {} + ); + + expect(Object.keys(merged.resources)).toEqual( + Object.keys(before.resources) + ); + expect(Object.keys(merged.functions)).toEqual( + Object.keys(before.functions) + ); + expect(Object.keys(merged.environment)).toEqual( + Object.keys(before.environment) + ); + }); + it('should create queue worker function', async () => { const appDefinition = { integrations: [ diff --git a/packages/devtools/infrastructure/domains/scheduler/scheduler-builder.js b/packages/devtools/infrastructure/domains/scheduler/scheduler-builder.js index daaf5c03b..63778e9b3 100644 --- a/packages/devtools/infrastructure/domains/scheduler/scheduler-builder.js +++ b/packages/devtools/infrastructure/domains/scheduler/scheduler-builder.js @@ -13,12 +13,24 @@ */ const { InfrastructureBuilder, ValidationResult } = require('../shared/base-builder'); +const { ResourceOwnership } = require('../shared/types/resource-ownership'); const { isScopedEnvironmentActive, getIntegrationFunctionNames, getAdminFunctionNames, } = require('../shared/function-environments'); +const queueArnFromUrl = (queueUrl) => { + const { hostname, pathname } = new URL(queueUrl); + const [accountId, queueName] = pathname.split('/').filter(Boolean); + return `arn:aws:sqs:${hostname.split('.')[1]}:${accountId}:${queueName}`; +}; + +const declaresRateLimit = (integration) => + Object.values(integration?.Definition?.modules ?? {}).some((module) => + Boolean(module?.definition?.API?.rateLimit) + ); + class SchedulerBuilder extends InfrastructureBuilder { constructor() { super(); @@ -37,7 +49,8 @@ class SchedulerBuilder extends InfrastructureBuilder { return appDefinition.integrations.some( (integration) => integration?.Definition?.webhooks?.enabled === true || - integration?.Definition?.webhooks === true + integration?.Definition?.webhooks === true || + declaresRateLimit(integration) ); } @@ -107,7 +120,11 @@ class SchedulerBuilder extends InfrastructureBuilder { if (Array.isArray(appDefinition.integrations)) { appDefinition.integrations.forEach((integration) => { const integrationName = integration?.Definition?.name; - if (integrationName) { + if ( + integration.ownership?.queue === ResourceOwnership.EXTERNAL + ) { + queueArns.push(queueArnFromUrl(integration.queue.url)); + } else if (integrationName) { const capitalizedName = integrationName.charAt(0).toUpperCase() + integrationName.slice(1); diff --git a/packages/devtools/infrastructure/domains/scheduler/scheduler-builder.test.js b/packages/devtools/infrastructure/domains/scheduler/scheduler-builder.test.js index 7dfeeda36..8adf1b0de 100644 --- a/packages/devtools/infrastructure/domains/scheduler/scheduler-builder.test.js +++ b/packages/devtools/infrastructure/domains/scheduler/scheduler-builder.test.js @@ -38,6 +38,61 @@ describe('SchedulerBuilder', () => { ).toBe(true); }); + it('runs when any module declares a rate limit', () => { + class Api { + static rateLimit = { minRetryAfterMs: 1_000 }; + } + expect( + schedulerBuilder.shouldExecute({ + integrations: [ + { Definition: { name: 'a' } }, + { + Definition: { + name: 'b', + modules: { crm: { definition: { API: Api } } }, + }, + }, + ], + }) + ).toBe(true); + }); + + it('skips a module whose API declares no rate limit', () => { + class Api {} + expect( + schedulerBuilder.shouldExecute({ + integrations: [ + { + Definition: { + name: 'a', + modules: { crm: { definition: { API: Api } } }, + }, + }, + ], + }) + ).toBe(false); + }); + + it.each([ + ['no modules', { name: 'a' }], + ['an empty modules object', { name: 'a', modules: {} }], + [ + 'a module with no definition', + { name: 'a', modules: { crm: {} } }, + ], + ['a null module', { name: 'a', modules: { crm: null } }], + [ + 'a definition with no API', + { name: 'a', modules: { crm: { definition: {} } } }, + ], + ])('skips an integration with %s', (_label, Definition) => { + expect( + schedulerBuilder.shouldExecute({ + integrations: [{ Definition }], + }) + ).toBe(false); + }); + it('skips otherwise', () => { expect( schedulerBuilder.shouldExecute({ @@ -115,4 +170,48 @@ describe('SchedulerBuilder', () => { expect(result.functionEnvironments).toBeUndefined(); }); }); + + describe('build() execution role', () => { + const sendableQueues = (result) => + result.resources.SchedulerExecutionRole.Properties.Policies[0] + .PolicyDocument.Statement[0].Resource; + + it('allows sending to a queue the stack creates', async () => { + const result = await schedulerBuilder.build( + { + scheduler: { enable: true }, + integrations: [{ Definition: { name: 'hubspot' } }], + }, + {} + ); + + expect(sendableQueues(result)).toEqual([ + { 'Fn::GetAtt': ['HubspotQueue', 'Arn'] }, + ]); + }); + + it('allows sending to an external queue by its ARN, not a resource the stack lacks', async () => { + const result = await schedulerBuilder.build( + { + scheduler: { enable: true }, + integrations: [ + { Definition: { name: 'hubspot' } }, + { + Definition: { name: 'salesforce' }, + ownership: { queue: 'external' }, + queue: { + url: 'https://sqs.us-east-1.amazonaws.com/123456789012/acme-salesforce', + }, + }, + ], + }, + {} + ); + + expect(sendableQueues(result)).toEqual([ + { 'Fn::GetAtt': ['HubspotQueue', 'Arn'] }, + 'arn:aws:sqs:us-east-1:123456789012:acme-salesforce', + ]); + }); + }); }); diff --git a/packages/devtools/infrastructure/domains/shared/utilities/base-definition-factory.js b/packages/devtools/infrastructure/domains/shared/utilities/base-definition-factory.js index fb4776ff2..4aafa0756 100644 --- a/packages/devtools/infrastructure/domains/shared/utilities/base-definition-factory.js +++ b/packages/devtools/infrastructure/domains/shared/utilities/base-definition-factory.js @@ -195,6 +195,7 @@ function createBaseDefinition( 'sqs:SendMessageBatch', 'sqs:GetQueueUrl', 'sqs:GetQueueAttributes', + 'sqs:ChangeMessageVisibility', ], Resource: [ { 'Fn::GetAtt': ['InternalErrorQueue', 'Arn'] }, diff --git a/packages/devtools/infrastructure/domains/shared/utilities/base-definition-factory.test.js b/packages/devtools/infrastructure/domains/shared/utilities/base-definition-factory.test.js index 8fcb111fd..fd3f0a366 100644 --- a/packages/devtools/infrastructure/domains/shared/utilities/base-definition-factory.test.js +++ b/packages/devtools/infrastructure/domains/shared/utilities/base-definition-factory.test.js @@ -135,6 +135,18 @@ describe('Base Definition Factory', () => { expect(sqsPermission).toBeDefined(); }); + it('should let a function extend the visibility timeout of a message on an integration queue', () => { + const result = createBaseDefinition({}, {}, {}); + + const sqsPermission = result.provider.iamRoleStatements.find( + (stmt) => stmt.Action.includes('sqs:SendMessage') + ); + expect(sqsPermission.Action).toContain( + 'sqs:ChangeMessageVisibility' + ); + expect(JSON.stringify(sqsPermission.Resource)).toContain('*Queue'); + }); + it('should include required plugins', () => { const result = createBaseDefinition({}, {}, {});