65155fe50c
Closes twentyhq/core-team-issues#2742 A logic function run is capped by its own `timeoutSeconds` (900s max), so anything that can't finish in one run — a full re-sync, a per-record fan-out, a rate-limited third-party API — had no way to continue. This adds a way to hand that work to the workers. ## What it looks like for an app author ```ts import { enqueueJob } from 'twenty-sdk/logic-function'; await enqueueJob({ logicFunctionUniversalIdentifier: '9f1c3d7e-51b8-4a29-8f0d-7c4e2a6b1d33', payload: { cursor: nextCursor }, retryLimit: 3, priority: 2, delayMs: 60_000, }); ``` The target runs in its own process with its own timeout budget. The classic shape is a function that enqueues *itself* with the next cursor until there is nothing left. ## Changes **twenty-shared** — `EnqueueJobInput` / `EnqueueJobOptions` / `EnqueueJobResult` in `application`. **twenty-server** — new `application-job` module under `core-modules/application`, following the `application-key-value` pattern: - `enqueueJob` mutation on the metadata API, `@AuthApplication`-scoped - the lookup is scoped to `applicationId` + `workspaceId` — that's the authorization boundary, an app can only enqueue its own logic functions, anything else is `LOGIC_FUNCTION_NOT_FOUND` - pushes a `LogicFunctionTriggerJob` onto the existing `logicFunctionQueue`, so the enqueued run goes through the same executor (and the same execution throttling) as every other trigger - the queued run inherits the caller's `userId`/`userWorkspaceId`, so its app access token carries the same permissions as the function that queued it **Job options** are range-checked via `ResolverValidationPipe`, since the values come from application code and an unbounded delay or retry count would let an app pin work in the shared queue: | Option | Default | Range | |--------|---------|-------| | `retryLimit` | `0` | `0`–`10` | | `priority` | queue default | `1`–`10` (lower first) | | `delayMs` | `0` | `0`–7 days | `retryLimit` defaults to `0` rather than inheriting the server-route path's `3`: retries re-run the whole handler, so opting in should be the author's explicit choice. **twenty-sdk** — `enqueueJob` in `twenty-sdk/logic-function`, same shape as `runAgent`/`kv`. **Docs** — new "Background Jobs" page under Extend → Apps → Logic, plus nav and overview entries. **Generated** — regenerated `twenty-front/src/generated-metadata` and `twenty-client-sdk/src/metadata/generated` for the new mutation. ## Tests - `application-job.service.spec.ts` — 5 unit tests: job options mapping, defaults, acting-user propagation, application-scoped lookup, not-found - `enqueue-job.integration-spec.ts` — 5 integration tests: rejects a non-`APPLICATION_ACCESS` token, enqueues a function the app owns, rejects a function owned by another application, rejects an unknown identifier, rejects out-of-range options All green locally, along with `typecheck` for `twenty-server`/`twenty-sdk` and oxlint/oxfmt on the touched files. ## Notes for review - The target is addressed by `universalIdentifier`, matching `runAgent({ agentUniversalIdentifier })` and `ServerRouteDispatchResult.targetLogicFunctionUniversalIdentifier`. Addressing by `name` would be friendlier, but logic function names aren't validated for uniqueness within an app — happy to add it as a convenience if you'd rather. - `enqueueJob` returns as soon as the job is accepted; it can't return the target's result, since the queue driver's `add` returns void. Documented, with a pointer to the KV store for handing results back. --- _Generated by [Claude Code](https://claude.ai/code/session_01QrYvGonS3HMdeuMAVjs5hR)_ <!-- This is an auto-generated description by cubic. --> <a href="https://cubic.dev/pr/twentyhq/twenty/pull/23527?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. --> --------- Co-authored-by: github-actions <github-actions@twenty.com>
166 lines
6.8 KiB
Plaintext
166 lines
6.8 KiB
Plaintext
---
|
||
title: Background Jobs
|
||
description: Hand long or rate-limited work to the Twenty workers by enqueuing another logic function run instead of doing everything inline.
|
||
icon: "layer-group"
|
||
---
|
||
|
||
A logic function run is capped by its `timeoutSeconds` (900 seconds maximum). Anything that can't finish in that window — a full re-sync, a per-record fan-out, a third-party API that rate-limits you — has to be split into smaller runs.
|
||
|
||
`enqueueJob` does exactly that: it asks the Twenty workers to run one of your app's logic functions later, in its own process, with its own timeout budget. The caller returns immediately.
|
||
|
||
```text
|
||
┌─────────────────┐ enqueueJob(...) ┌──────────────┐ ┌────────────────────┐
|
||
│ Logic function │ ─────────────────▶ │ Job queue │──▶│ Logic function │
|
||
│ (returns now) │ │ (workers) │ │ (fresh run/timeout)│
|
||
└─────────────────┘ └──────────────┘ └────────────────────┘
|
||
```
|
||
|
||
## Enqueue a run
|
||
|
||
Import `enqueueJob` from `twenty-sdk/logic-function` and point it at the `universalIdentifier` of the logic function you want to run.
|
||
|
||
```ts src/logic-functions/sync-all-contacts.ts
|
||
import { enqueueJob } from 'twenty-sdk/logic-function';
|
||
|
||
await enqueueJob({
|
||
logicFunctionUniversalIdentifier: '9f1c3d7e-51b8-4a29-8f0d-7c4e2a6b1d33',
|
||
payload: { page: 1 },
|
||
});
|
||
```
|
||
|
||
The target function receives `payload` as its handler argument, exactly like any other trigger. It must belong to the **same application** as the caller — enqueuing another app's function is rejected with `Logic function not found`.
|
||
|
||
<Note>
|
||
`enqueueJob` returns as soon as the job is accepted, not when it has run. It does not return the target's result — have the target write what it produces to the [key-value store](/developers/extend/apps/logic/key-value-store) or to a workspace record if you need to read it back.
|
||
</Note>
|
||
|
||
## Job options
|
||
|
||
| Option | Default | Range | What it does |
|
||
|--------|---------|-------|--------------|
|
||
| `retryLimit` | `0` | `0`–`10` | Extra attempts if the run throws. Only raise this for handlers that are safe to run twice. |
|
||
| `delayMs` | `0` | `0`–`604800000` (7 days) | Wait this long before the run becomes eligible. |
|
||
|
||
```ts
|
||
await enqueueJob({
|
||
logicFunctionUniversalIdentifier: '9f1c3d7e-51b8-4a29-8f0d-7c4e2a6b1d33',
|
||
payload: { page: 1 },
|
||
retryLimit: 3,
|
||
delayMs: 60_000,
|
||
});
|
||
```
|
||
|
||
<Note>
|
||
**Priority is not configurable yet.** Enqueued jobs always run at the lowest priority, so platform work is never delayed behind application jobs. Control over priority is coming soon.
|
||
</Note>
|
||
|
||
The queued run inherits the acting user of the function that enqueued it, so it acts with the same permissions.
|
||
|
||
## Use it: page through a long sync
|
||
|
||
The classic shape is a function that enqueues *itself* with the next cursor. Each run does one page of work well inside its own timeout, and the chain stops when there is nothing left.
|
||
|
||
```ts src/logic-functions/sync-contacts-page.ts
|
||
import { defineLogicFunction } from 'twenty-sdk/define';
|
||
import { enqueueJob } from 'twenty-sdk/logic-function';
|
||
|
||
const SYNC_CONTACTS_PAGE = '9f1c3d7e-51b8-4a29-8f0d-7c4e2a6b1d33';
|
||
|
||
const handler = async (params: { cursor?: string }) => {
|
||
const { contacts, nextCursor } = await fetchContactsPage(params.cursor);
|
||
|
||
await importContacts(contacts);
|
||
|
||
if (nextCursor) {
|
||
await enqueueJob({
|
||
logicFunctionUniversalIdentifier: SYNC_CONTACTS_PAGE,
|
||
payload: { cursor: nextCursor },
|
||
delayMs: 2_000,
|
||
});
|
||
}
|
||
|
||
return { imported: contacts.length, done: !nextCursor };
|
||
};
|
||
|
||
export default defineLogicFunction({
|
||
universalIdentifier: SYNC_CONTACTS_PAGE,
|
||
name: 'sync-contacts-page',
|
||
timeoutSeconds: 120,
|
||
handler,
|
||
});
|
||
```
|
||
|
||
## Fan out per record
|
||
|
||
When the work is naturally per-item, enqueue one job per item and let the workers process them in parallel instead of looping inline.
|
||
|
||
```ts
|
||
const companies = await listCompaniesToEnrich();
|
||
|
||
await Promise.all(
|
||
companies.map((company) =>
|
||
enqueueJob({
|
||
logicFunctionUniversalIdentifier: ENRICH_COMPANY,
|
||
payload: { companyId: company.id },
|
||
retryLimit: 2,
|
||
}),
|
||
),
|
||
);
|
||
```
|
||
|
||
## Good practice for long-running work
|
||
|
||
Two rules cover almost every long job: **recurse instead of looping**, and **process a bounded chunk per run**.
|
||
|
||
A run that tries to do everything is the failure mode — it hits the timeout, and with a retry it starts the whole thing again from zero. Instead, size one chunk so it comfortably finishes inside `timeoutSeconds`, persist your position, and enqueue the next run.
|
||
|
||
```ts src/logic-functions/enrich-companies-batch.ts
|
||
import { defineLogicFunction } from 'twenty-sdk/define';
|
||
import { enqueueJob, kv } from 'twenty-sdk/logic-function';
|
||
|
||
const ENRICH_COMPANIES_BATCH = '3f9d1c02-8a44-4f0e-b1d7-9c2e5a7b4f10';
|
||
const CHUNK_SIZE = 50;
|
||
|
||
const handler = async (params: { offset?: number }) => {
|
||
const offset = params.offset ?? 0;
|
||
const companies = await listCompaniesToEnrich({
|
||
offset,
|
||
limit: CHUNK_SIZE,
|
||
});
|
||
|
||
for (const company of companies) {
|
||
await enrichCompany(company);
|
||
}
|
||
|
||
await kv.set('enrich:progress', { offset: offset + companies.length });
|
||
|
||
if (companies.length === CHUNK_SIZE) {
|
||
await enqueueJob({
|
||
logicFunctionUniversalIdentifier: ENRICH_COMPANIES_BATCH,
|
||
payload: { offset: offset + CHUNK_SIZE },
|
||
});
|
||
}
|
||
|
||
return { processed: companies.length, done: companies.length < CHUNK_SIZE };
|
||
};
|
||
|
||
export default defineLogicFunction({
|
||
universalIdentifier: ENRICH_COMPANIES_BATCH,
|
||
name: 'enrich-companies-batch',
|
||
timeoutSeconds: 300,
|
||
handler,
|
||
});
|
||
```
|
||
|
||
What makes this hold up:
|
||
|
||
- **Size the chunk from the slowest item, not the average.** `CHUNK_SIZE × worst-case item time` has to fit in `timeoutSeconds` with room to spare, or the tail of a chunk is lost when the run is cut off.
|
||
- **Make the terminating condition explicit.** Recurse only while a full chunk came back. A chain that stops on "no results" alone will keep going forever if the source ever returns a short page mid-way.
|
||
- **Persist progress before enqueuing the next run,** so a failed link restarts from the last completed chunk instead of the beginning.
|
||
- **Keep each chunk idempotent.** Reprocessing one chunk after a retry must not double-write — key writes on the record or external id you are processing.
|
||
- **Prefer a chunked chain over one giant fan-out** when the work hits a rate-limited third party: a chain with `delayMs` paces itself, whereas thousands of jobs enqueued at once all become eligible immediately.
|
||
|
||
<Warning>
|
||
Retries re-run the whole handler. Keep enqueued handlers idempotent before setting `retryLimit` above `0`.
|
||
</Warning>
|