Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
34 commits
Select commit Hold shift + click to select a range
ef792e4
feat(core): add the constants and counters for deferred queue messages
d-klotz Sep 28, 2026
288b760
feat(core): keep the rate limit on a failed token refresh
d-klotz Sep 28, 2026
f9c3b5d
feat(core): put a rate-limited queue message back at its retry time
d-klotz Sep 28, 2026
e0b75dd
feat(core): write the rate-limit wait on the process of a deferred me…
d-klotz Sep 28, 2026
3518b14
feat(devtools): provision what a deferred queue message needs
d-klotz Sep 28, 2026
7bafa35
docs(core): document the rate-limit deferral in the queue worker
d-klotz Sep 28, 2026
ebe5c60
feat(core): type the Worker rate-limit hooks
d-klotz Sep 28, 2026
8f335eb
Merge remote-tracking branch 'origin/feat/core-rate-limit-requester' …
d-klotz Sep 28, 2026
b14221f
fix(core): address the review of the queue worker deferral
d-klotz Sep 28, 2026
3df9ca1
refactor(core): drop the comments of the rate-limit deferral
d-klotz Sep 29, 2026
d20b32f
refactor(devtools): keep the scheduler builder comment as it was
d-klotz Sep 29, 2026
4536c17
test(core): build the deferred message body of the Worker tests in on…
d-klotz Sep 29, 2026
897a1f7
feat(core): allow 30 deferrals over 26 hours by default
d-klotz Sep 29, 2026
1df4ed1
Merge feat/core-rate-limit-requester into feat/core-rate-limit-defer
d-klotz Sep 29, 2026
208336f
refactor(core): define toPositiveInteger once for the queue modules
d-klotz Sep 29, 2026
b24ecef
fix(core): keep the visibility timeout under the 12 hours that SQS co…
d-klotz Sep 29, 2026
6a66f4e
refactor(core): pass one deferral object through the deferral tiers
d-klotz Sep 29, 2026
efd2dcf
fix(core): write one log record for each deferral outcome
d-klotz Sep 29, 2026
6e10d41
fix(core): log the cause of a failed deferral as an error
d-klotz Sep 29, 2026
e4d6dec
fix(core): treat a scheduler that cannot be built like a failed schedule
d-klotz Sep 29, 2026
c01804e
refactor(core): move the rate-limit run state into two use cases
d-klotz Sep 29, 2026
775b1b8
test(core): clear the rate-limit run state after a deferred message runs
d-klotz Sep 29, 2026
910d5a8
docs(core): describe the deferral caps, the visibility margin and the…
d-klotz Sep 29, 2026
5575ec4
fix(core): keep the module name on a rate-limited token refresh error
d-klotz Sep 29, 2026
4b25a44
test(devtools): pin the template of an app that declares no rate limit
d-klotz Sep 29, 2026
6a81b60
chore(core): declare aws-sdk-client-mock as a dev dependency
d-klotz Sep 29, 2026
07aea31
fix(core): point the scheduler client at the local endpoint in a loca…
d-klotz Sep 30, 2026
e372ef6
fix(core): let the message that set a rate-limit wait clear it at once
d-klotz Sep 30, 2026
e56cf82
Merge feat/core-rate-limit-requester into feat/core-rate-limit-defer
d-klotz Nov 9, 2026
3c0bea9
test(core): pin the Worker for a RateLimitError with no retry time
d-klotz Nov 9, 2026
18fb0ee
docs(core): say a RateLimitError with no retryAt returns to SQS
d-klotz Nov 9, 2026
798cee9
fix(core): write the rate-limit wait before the delayed send
d-klotz Nov 9, 2026
c1ce60b
Merge branch 'feat/core-rate-limit-requester' into feat/core-rate-lim…
d-klotz Nov 9, 2026
f16d20b
fix(devtools): let the scheduler send to an external integration queue
d-klotz Nov 9, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions .claude/skills/frigg-scheduled-jobs/SKILL.md
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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.
2 changes: 1 addition & 1 deletion .claude/skills/frigg/references/infrastructure.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
39 changes: 36 additions & 3 deletions docs/guides/INTEGRATION-PATTERNS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:

Expand Down
13 changes: 12 additions & 1 deletion docs/guides/LOGGING-IN-INTEGRATIONS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand Down
6 changes: 4 additions & 2 deletions docs/reference/api-module-definition-and-functions.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
2 changes: 2 additions & 0 deletions package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

88 changes: 81 additions & 7 deletions packages/core/CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
}
Expand All @@ -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.
Expand Down Expand Up @@ -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

Expand Down
30 changes: 30 additions & 0 deletions packages/core/core/CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading
Loading