houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix Cloudflare rate limiting during last-contact backfill (#22811)

Browse Source 
## Context

Upgrading `twenty-last-contact` on a production workspace failed with a
Cloudflare error 1015 ("You are being rate limited"). The
`backfill-last-contact` post-install function runs on every version
upgrade and fired 20 concurrent update mutations per batch with no pause
between batches, on top of paginated full-collection reads. On a
workspace with real email/calendar history that burst trips Cloudflare's
rate limit, and since the client SDK throws on any non-2xx response, a
single 429 killed the whole install/upgrade hook mid-backfill.

## Changes

- New `executeWithRetry` util: retries rate-limit (429 / Cloudflare
1015) and transient gateway/network errors (502/503/504, timeouts,
connection resets) with exponential backoff and jitter, capped at 5
attempts. Honors a `retry_after` hint when present in the response body.
Non-retryable errors still throw immediately.
- All backfill queries and mutations are wrapped with it.
- Update batch concurrency reduced from 20 to 10 to keep bursts under
the rate limit in the first place.
- Bumped app version to 1.1.1 with a changelog entry.

## Test

- Added unit tests for `executeWithRetry` (success passthrough,
retry-then-succeed, non-retryable passthrough, retry exhaustion,
`retry_after` handling).
- `yarn test:unit` (28 passed), `yarn typecheck`, `yarn lint` all green
in the app package.

---
_Generated by [Claude
Code](https://claude.ai/code/session_01AtnkEfbpFhLp5qCJbSmZpE)_

<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/22811?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->
This commit is contained in:
martmull
2026-07-10 15:26:07 +02:00
committed by GitHub
parent 7861ada589
commit 7f8a1da27a
5 changed files with 234 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-apps/public/twenty-last-contact/CHANGELOG.md
+4
View File
@@ -1,5 +1,9 @@
# Changelog
## 1.1.1
- Throttle backfill updates and retry rate-limited or transient API failures with exponential backoff, so install/upgrade no longer fails behind Cloudflare rate limiting.
## 1.1.0
- Add last contact on Companies and Opportunities.
packages/twenty-apps/public/twenty-last-contact/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@twentyhq/last-contact",
"version": "1.1.0",
"version": "1.1.1",
"license": "MIT",
"engines": {
"node": "^24.5.0",
packages/twenty-apps/public/twenty-last-contact/src/logic-functions/backfill-last-contact.ts
+104 -86
View File
@@ -2,9 +2,13 @@ import { definePostInstallLogicFunction } from 'twenty-sdk/define';
import { CoreApiClient } from 'twenty-client-sdk/core';
import { BACKFILL_POST_INSTALL_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER } from 'src/constants/universal-identifiers';
import { executeWithRetry } from 'src/utils/execute-with-retry';
const PAGE_SIZE = 200;
const UPDATE_BATCH_SIZE = 20;
// Kept low so update bursts stay under Cloudflare rate limiting on hosted
// workspaces; executeWithRetry absorbs the occasional 429 that still slips
// through.
const UPDATE_BATCH_SIZE = 10;
type EmailInteraction = {
personId: string;
@@ -53,23 +57,25 @@ const collectEmailInteractions = async (
let after: string | undefined;
do {
const { messageParticipants } = await client.query({
messageParticipants: {
__args: {
filter: { personId: { is: 'NOT_NULL' } },
first: PAGE_SIZE,
after,
},
edges: {
node: {
id: true,
personId: true,
message: { id: true, receivedAt: true },
const { messageParticipants } = await executeWithRetry(() =>
client.query({
messageParticipants: {
__args: {
filter: { personId: { is: 'NOT_NULL' } },
first: PAGE_SIZE,
after,
},
edges: {
node: {
id: true,
personId: true,
message: { id: true, receivedAt: true },
},
},
pageInfo: { hasNextPage: true, endCursor: true },
},
pageInfo: { hasNextPage: true, endCursor: true },
},
});
}),
);
for (const edge of messageParticipants?.edges ?? []) {
const { personId, message } = edge.node;
@@ -98,23 +104,25 @@ const collectMeetingInteractions = async (
let after: string | undefined;
do {
const { calendarEventParticipants } = await client.query({
calendarEventParticipants: {
__args: {
filter: { personId: { is: 'NOT_NULL' } },
first: PAGE_SIZE,
after,
},
edges: {
node: {
id: true,
personId: true,
calendarEvent: { id: true, startsAt: true, isCanceled: true },
const { calendarEventParticipants } = await executeWithRetry(() =>
client.query({
calendarEventParticipants: {
__args: {
filter: { personId: { is: 'NOT_NULL' } },
first: PAGE_SIZE,
after,
},
edges: {
node: {
id: true,
personId: true,
calendarEvent: { id: true, startsAt: true, isCanceled: true },
},
},
pageInfo: { hasNextPage: true, endCursor: true },
},
pageInfo: { hasNextPage: true, endCursor: true },
},
});
}),
);
for (const edge of calendarEventParticipants?.edges ?? []) {
const { personId, calendarEvent } = edge.node;
@@ -151,22 +159,24 @@ const collectMessageMemberInfo = async (
let after: string | undefined;
do {
const { messageParticipants } = await client.query({
messageParticipants: {
__args: {
filter: {
messageId: { in: ids },
workspaceMemberId: { is: 'NOT_NULL' },
const { messageParticipants } = await executeWithRetry(() =>
client.query({
messageParticipants: {
__args: {
filter: {
messageId: { in: ids },
workspaceMemberId: { is: 'NOT_NULL' },
},
first: PAGE_SIZE,
after,
},
first: PAGE_SIZE,
after,
edges: {
node: { messageId: true, role: true, workspaceMemberId: true },
},
pageInfo: { hasNextPage: true, endCursor: true },
},
edges: {
node: { messageId: true, role: true, workspaceMemberId: true },
},
pageInfo: { hasNextPage: true, endCursor: true },
},
});
}),
);
for (const edge of messageParticipants?.edges ?? []) {
const { messageId, role, workspaceMemberId } = edge.node;
@@ -203,26 +213,28 @@ const collectCalendarOwners = async (
let after: string | undefined;
do {
const { calendarEventParticipants } = await client.query({
calendarEventParticipants: {
__args: {
filter: {
calendarEventId: { in: ids },
workspaceMemberId: { is: 'NOT_NULL' },
const { calendarEventParticipants } = await executeWithRetry(() =>
client.query({
calendarEventParticipants: {
__args: {
filter: {
calendarEventId: { in: ids },
workspaceMemberId: { is: 'NOT_NULL' },
},
first: PAGE_SIZE,
after,
},
first: PAGE_SIZE,
after,
},
edges: {
node: {
calendarEventId: true,
isOrganizer: true,
workspaceMemberId: true,
edges: {
node: {
calendarEventId: true,
isOrganizer: true,
workspaceMemberId: true,
},
},
pageInfo: { hasNextPage: true, endCursor: true },
},
pageInfo: { hasNextPage: true, endCursor: true },
},
});
}),
);
for (const edge of calendarEventParticipants?.edges ?? []) {
const { calendarEventId, isOrganizer, workspaceMemberId } = edge.node;
@@ -251,17 +263,19 @@ const collectPersonCompanies = async (
let after: string | undefined;
do {
const { people } = await client.query({
people: {
__args: {
filter: { companyId: { is: 'NOT_NULL' } },
first: PAGE_SIZE,
after,
const { people } = await executeWithRetry(() =>
client.query({
people: {
__args: {
filter: { companyId: { is: 'NOT_NULL' } },
first: PAGE_SIZE,
after,
},
edges: { node: { id: true, companyId: true } },
pageInfo: { hasNextPage: true, endCursor: true },
},
edges: { node: { id: true, companyId: true } },
pageInfo: { hasNextPage: true, endCursor: true },
},
});
}),
);
for (const edge of people?.edges ?? []) {
const { id, companyId } = edge.node;
@@ -285,15 +299,17 @@ const collectOpportunities = async (
let after: string | undefined;
do {
const { opportunities: page } = await client.query({
opportunities: {
__args: { first: PAGE_SIZE, after },
edges: {
node: { id: true, pointOfContactId: true },
const { opportunities: page } = await executeWithRetry(() =>
client.query({
opportunities: {
__args: { first: PAGE_SIZE, after },
edges: {
node: { id: true, pointOfContactId: true },
},
pageInfo: { hasNextPage: true, endCursor: true },
},
pageInfo: { hasNextPage: true, endCursor: true },
},
});
}),
);
for (const edge of page?.edges ?? []) {
const { id, pointOfContactId } = edge.node;
@@ -401,12 +417,14 @@ const applyUpdates = async (
for (const batch of chunk(updates, UPDATE_BATCH_SIZE)) {
await Promise.all(
batch.map(({ id, data }) =>
client.mutation({
[mutationName]: {
__args: { id, data },
id: true,
},
}),
executeWithRetry(() =>
client.mutation({
[mutationName]: {
__args: { id, data },
id: true,
},
}),
),
),
);
}
packages/twenty-apps/public/twenty-last-contact/src/utils/__tests__/execute-with-retry.test.ts
+74
View File
@@ -0,0 +1,74 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { executeWithRetry } from 'src/utils/execute-with-retry';
beforeEach(() => {
vi.useFakeTimers();
vi.spyOn(Math, 'random').mockReturnValue(0);
});
afterEach(() => {
vi.useRealTimers();
vi.restoreAllMocks();
});
describe('executeWithRetry', () => {
it('returns the result on first success', async () => {
const execute = vi.fn().mockResolvedValue('ok');
await expect(executeWithRetry(execute)).resolves.toBe('ok');
expect(execute).toHaveBeenCalledTimes(1);
});
it('retries rate-limited requests until they succeed', async () => {
const execute = vi
.fn()
.mockRejectedValueOnce(new Error('Too Many Requests: error code 1015'))
.mockResolvedValue('ok');
const promise = executeWithRetry(execute);
await vi.advanceTimersByTimeAsync(2_000);
await expect(promise).resolves.toBe('ok');
expect(execute).toHaveBeenCalledTimes(2);
});
it('does not retry non-retryable errors', async () => {
const execute = vi
.fn()
.mockRejectedValue(new Error('Bad Request: invalid data'));
await expect(executeWithRetry(execute)).rejects.toThrow('Bad Request');
expect(execute).toHaveBeenCalledTimes(1);
});
it('gives up after exhausting retries', async () => {
const execute = vi
.fn()
.mockRejectedValue(new Error('Gateway time-out: origin overloaded'));
const promise = executeWithRetry(execute);
const assertion = expect(promise).rejects.toThrow('Gateway time-out');
await vi.advanceTimersByTimeAsync(60_000);
await assertion;
expect(execute).toHaveBeenCalledTimes(5);
});
it('waits for the server-provided retry_after when it exceeds the backoff', async () => {
const execute = vi
.fn()
.mockRejectedValueOnce(
new Error('Gateway time-out: {"retryable": true, "retry_after": 10}'),
)
.mockResolvedValue('ok');
const promise = executeWithRetry(execute);
await vi.advanceTimersByTimeAsync(9_000);
expect(execute).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1_000);
await expect(promise).resolves.toBe('ok');
expect(execute).toHaveBeenCalledTimes(2);
});
});
packages/twenty-apps/public/twenty-last-contact/src/utils/execute-with-retry.ts
+51
View File
@@ -0,0 +1,51 @@
const MAX_ATTEMPTS = 5;
const INITIAL_RETRY_DELAY_MS = 2_000;
const MAX_RETRY_DELAY_MS = 30_000;
const MAX_JITTER_MS = 1_000;
// The client SDK surfaces HTTP failures as plain Error messages built from the
// status text and raw response body, so retryability has to be detected from
// the message text. Covers rate limiting (429, Cloudflare 1015), transient
// gateway errors (502/503/504) and network-level failures.
const RETRYABLE_ERROR_PATTERN =
/\b(429|1015|too many requests|rate ?limit\w*|502|503|504|bad gateway|gateway time-?out|service unavailable|timed? ?out|fetch failed|econnreset|econnrefused|socket hang up)\b/i;
const sleep = (durationMs: number): Promise<void> =>
new Promise((resolve) => setTimeout(resolve, durationMs));
const getErrorMessage = (error: unknown): string =>
error instanceof Error ? error.message : String(error);
const isRetryableError = (error: unknown): boolean =>
RETRYABLE_ERROR_PATTERN.test(getErrorMessage(error));
const parseRetryAfterMs = (error: unknown): number | undefined => {
const match = getErrorMessage(error).match(/"retry_after"\s*:\s*(\d+)/);
return match ? Number(match[1]) * 1_000 : undefined;
};
export const executeWithRetry = async <TResult>(
execute: () => TResult,
): Promise<Awaited<TResult>> => {
for (let attempt = 1; ; attempt += 1) {
try {
return await execute();
} catch (error) {
if (attempt >= MAX_ATTEMPTS || !isRetryableError(error)) {
throw error;
}
const backoffMs = Math.min(
INITIAL_RETRY_DELAY_MS * 2 ** (attempt - 1),
MAX_RETRY_DELAY_MS,
);
const retryAfterMs = Math.min(
parseRetryAfterMs(error) ?? 0,
MAX_RETRY_DELAY_MS,
);
const jitterMs = Math.random() * MAX_JITTER_MS;
await sleep(Math.max(backoffMs, retryAfterMs) + jitterMs);
}
}
};