Sarcini în fundal

Prezentare generală

O rulare a unei funcții de logică este limitată de timeoutSeconds (maximum 900 de secunde). Orice lucru care nu poate fi finalizat în acel interval — o resincronizare completă, un fan-out per înregistrare, un API de la o terță parte care aplică limitare de rată — trebuie împărțit în execuții mai mici. enqueueJobs face exact asta: le cere lucrătorilor Twenty să ruleze mai târziu una dintre funcțiile de logică ale aplicației tale, o dată pentru fiecare payload, fiecare rulare în propriul său proces, cu propriul său buget de timeout. Apelantul returnează imediat.

  ┌─────────────────┐  enqueueJobs(...)  ┌──────────────┐   ┌────────────────────┐
  │ Logic function  │ ─────────────────▶ │ Job queue    │──▶│ Logic function     │
  │ (returns now)   │                    │ (workers)    │   │ (fresh run/timeout)│
  └─────────────────┘                    └──────────────┘   └────────────────────┘

Pune rulările în coadă

Importă enqueueJobs din twenty-sdk/logic-function, indică-l către universalIdentifier al funcției logice de rulat și transmite câte un payload pentru fiecare rulare funcției enqueue.

src/logic-functions/sync-all-contacts.ts

import { enqueueJobs } from 'twenty-sdk/logic-function';

await enqueueJobs({
  logicFunctionUniversalIdentifier: '9f1c3d7e-51b8-4a29-8f0d-7c4e2a6b1d33',
  jobs: [{ payload: { page: 1 } }],
});

Fiecare rulare primește payloadul său ca argument al handlerului, exact ca orice alt declanșator. Ținta trebuie să aparțină aceleiași aplicații ca apelantul — punerea în coadă a funcției altei aplicații este respinsă cu Logic function not found și nu se pune nimic în coadă. Un singur apel acceptă până la 200 joburi.

enqueueJobs revine imediat ce joburile sunt acceptate, nu când au fost rulate. Nu întoarce rezultatele țintelor — fă ca fiecare țintă să scrie ce produce în magazinul cheie-valoare sau într-o înregistrare din spațiul de lucru dacă trebuie să le citești ulterior. Pentru a afla dacă o execuție s-a încheiat, interogați periodic starea jobului.

Ajutorul mai vechi enqueueJob, care pune în coadă un singur job per apel, este depreciat. Folosiți în schimb enqueueJobs cu o listă jobs cu un singur element. Argumentul payloads este și el învechit — transmiteți în schimb jobs, care acceptă aceeași încărcătură utilă pentru fiecare execuție și vă permite să denumiți fiecare execuție.

Interogați periodic joburile

enqueueJobs returnează câte un jobId pentru fiecare execuție pusă în coadă, în aceeași ordine ca lucrările pe care le-ai transmis. getJobs citește stadiul la care au ajuns acele execuții, într-un singur apel.

import { enqueueJobs, getJobs } from 'twenty-sdk/logic-function';

const { jobIds } = await enqueueJobs({
  logicFunctionUniversalIdentifier: ENRICH_COMPANY,
  jobs: [{ payload: { companyId: 'abc' } }],
});

const statuses = await getJobs(jobIds);

Un singur apel getJobs citește cel mult 200 de ID-uri, aceeași limită pe care enqueueJobs o aplică unui lot. Fiecare intrare conține ID-ul solicitat:

Câmp	Ce conține
`jobId`	ID-ul interogat.
`state`	`WAITING`, `PRIORITIZED`, `DELAYED`, `ACTIVE`, `WAITING_CHILDREN`, `COMPLETED` sau `FAILED`.
`attemptsMade`	Încercările din coadă efectuate până acum, inclusiv cea curentă.
`failedReason`	Mesajul de eroare al ultimei încercări eșuate sau, în caz contrar, `null`.
`enqueuedAt`	Momentul în care lucrarea a fost acceptată, în milisecunde epoch.
`startedAt`	Momentul în care un lucrător a preluat-o; `null` cât timp încă așteaptă.
`finishedAt`	Momentul în care s-a finalizat sau a eșuat; `null` cât timp încă rulează.

O lucrare care nu a fost încă preluată apare în mod normal ca PRIORITIZED, nu ca WAITING, deoarece lucrările aplicației sunt puse în coadă la o prioritate mică fixă. COMPLETED și FAILED sunt stările terminale. O buclă de interogare ar trebui, de asemenea, să renunțe singură după un număr limitat de încercări, în loc să aștepte la nesfârșit, deoarece o execuție poate fi citită doar cât timp coada o păstrează: 4 ore după finalizare, 7 zile după eșec și, în ambele cazuri, doar cele mai recente 1000 de lucrări în acea stare. getJobs returnează numai lucrările pe care încă le păstrează, astfel încât un ID care a fost eliminat — sau nu a existat niciodată — pur și simplu lipsește din rezultat, în loc să provoace o eroare pentru întregul lot. Compară ceea ce primești înapoi cu ceea ce ai solicitat pentru a le identifica și tratează un ID lipsă ca „nu mai este cunoscut”, ceea ce nu este același răspuns ca „nu a rulat niciodată”. Stările sunt limitate la spațiul de lucru din care se efectuează apelul, astfel încât un spațiu de lucru nu poate citi niciodată lucrarea altui spațiu de lucru.

Alege propriul ID al lucrării

Transmite jobs în loc de payloads pentru a denumi tu însuți fiecare execuție. ID-ul pe care îl transmiți este cel pe care îl acceptă getJobs, astfel încât un apelant poate interoga o lucrare fără a fi nevoit să stocheze ceea ce a returnat enqueueJobs.

await enqueueJobs({
  logicFunctionUniversalIdentifier: ENRICH_COMPANY,
  jobs: [{ payload: { companyId: 'abc' }, jobId: 'enrich-abc' }],
});

Un ID are cel mult 128 de caractere: litere, cifre, _, . sau -, iar două lucrări dintr-un singur apel nu pot avea același ID. Transmite fie payloads, fie jobs, niciodată ambele. Când omiți jobId, Twenty generează unul. Un jobId este, de asemenea, o cheie de idempotență: punerea în coadă a unui ID pe care coada îl păstrează încă este acceptată, dar nu pornește o a doua execuție, iar getJobs continuă să raporteze prima execuție. Astfel, reîncercarea punerii în coadă este sigură. Două avertismente: garanția durează numai cât timp lucrarea este păstrată, astfel încât același ID poate porni o execuție nouă după ce cea originală a fost eliminată, iar ID-urile sunt partajate de fiecare aplicație dintr-un spațiu de lucru, așa că adaugă un prefix ID-urilor tale dacă o coliziune ar conta.

Opțiuni job

Opțiunile se aplică fiecărei rulări din lot.

Opțiune	Implicit	Interval	Ce face
`retryLimit`	`0`	`0`–`10`	Numărul total de încercări suplimentare în coadă. Reîncercările solicitate de aplicație sunt limitate la `3`. Crește această valoare numai pentru handleri care sunt siguri de rulat de două ori.
`delayMs`	`0`	`0`–`604800000` (7 zile)	Așteaptă atât timp înainte ca rulările să devină eligibile.

await enqueueJobs({
  logicFunctionUniversalIdentifier: '9f1c3d7e-51b8-4a29-8f0d-7c4e2a6b1d33',
  jobs: [{ payload: { page: 1 } }],
  retryLimit: 3,
  delayMs: 60_000,
});

Prioritatea nu este încă configurabilă. Joburile puse în coadă rulează întotdeauna cu cea mai joasă prioritate, astfel încât activitatea platformei să nu fie niciodată întârziată de joburile aplicației. Controlul asupra priorității va fi disponibil în curând.

Rularea pusă în coadă moștenește utilizatorul activ al funcției care a pus-o în coadă, astfel încât acționează cu aceleași permisiuni.

Reîncercați după un eșec temporar

Twenty nu reîncearcă fiecare excepție din codul aplicației. O eroare obișnuită lansată este tratată ca un eșec permanent. Pentru un eșec temporar, o funcție logică pusă în coadă poate solicita până la trei reîncercări prin lansarea RetryableLogicFunctionError.

import {
  type LogicFunctionExecutionContext,
  RetryableLogicFunctionError,
} from 'twenty-sdk/logic-function';

export const handler = async (
  _payload: unknown,
  { retryCount, maxRetries }: LogicFunctionExecutionContext,
) => {
  const response = await fetch('https://api.example.com/contacts');

  if (response.status === 429 || response.status >= 500) {
    throw new RetryableLogicFunctionError(
      `The contacts API is temporarily unavailable (${response.status}); retry ${retryCount} of ${maxRetries}`,
    );
  }
};

Lansați direct RetryableLogicFunctionError atunci când este posibil. Dacă o extindeți, nu îi înlocuiți name: Twenty recunoaște numele serializat RetryableLogicFunctionError în toate mediile de execuție. retryCount este 0 pentru execuția inițială și crește doar atunci când codul aplicației solicită o reîncercare. maxRetries este cel mult 3 și poate fi mai mic atunci când sarcina din coadă are o limită totală de reîncercări mai mică. Eșecurile platformei nu cresc retryCount, deși consumă în continuare bugetul general de siguranță al cozii. Coada întârzie încercările de reîncercare folosind backoff exponențial și jitter. Întârzierea exactă nu este garantată în mod intenționat, astfel că codul aplicației nu ar trebui să depindă de producerea unei reîncercări la un moment precis. După ce se atinge maxRetries, o altă RetryableLogicFunctionError este înregistrată ca eșecul final al aplicației, fără o altă execuție.

Reîncercările rulează din nou întregul handler și se pot produce după ce unele efecte secundare au reușit. Faceți handlerul idempotent înainte de a solicita reîncercări.

Folosește-l: parcurge pe pagini o sincronizare lungă

Forma clasică este o funcție care pune în coadă ea însăși cu următorul cursor. Fiecare rulare procesează o pagină de lucru bine în interiorul propriului timeout, iar lanțul se oprește când nu mai rămâne nimic.

src/logic-functions/sync-contacts-page.ts

import { defineLogicFunction } from 'twenty-sdk/define';
import { enqueueJobs } 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 enqueueJobs({
      logicFunctionUniversalIdentifier: SYNC_CONTACTS_PAGE,
      jobs: [{ 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,
});

Dispersare pe înregistrare

Când munca este în mod natural pe element, pune în coadă un job per element într-un singur apel și lasă workerii să le proceseze în paralel în loc să iterezi inline.

const companies = await listCompaniesToEnrich();

await enqueueJobs({
  logicFunctionUniversalIdentifier: ENRICH_COMPANY,
  jobs: companies.map((company) => ({ payload: { companyId: company.id } })),
  retryLimit: 2,
});

Bune practici pentru muncă de lungă durată

Două reguli acoperă aproape orice job lung: folosește recursie în locul buclelor și procesează un segment limitat per rulare. O rulare care încearcă să facă totul este modul de eșec — atinge timeout-ul, iar la o reîncercare începe din nou totul de la zero. În schimb, dimensionează un segment astfel încât să se termine confortabil în interiorul timeoutSeconds, persistă-ți poziția și pune în coadă următoarea rulare.

src/logic-functions/enrich-companies-batch.ts

import { defineLogicFunction } from 'twenty-sdk/define';
import { enqueueJobs, 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 enqueueJobs({
      logicFunctionUniversalIdentifier: ENRICH_COMPANIES_BATCH,
      jobs: [{ 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,
});

Ce îl face robust:

Dimensionează segmentul pornind de la cel mai lent element, nu de la medie. CHUNK_SIZE × worst-case item time trebuie să încapă în timeoutSeconds cu marjă, altfel „coada” unui segment se pierde când rularea este întreruptă.
Fă condiția de terminare explicită. Apelează recursiv doar cât timp a revenit un segment complet. Un lanț care se oprește doar pe baza „niciun rezultat” va continua la nesfârșit dacă sursa returnează vreodată o pagină scurtă la mijloc.
Păstrează progresul înainte de a pune în coadă următoarea rulare, astfel încât o verigă eșuată să repornească de la ultimul segment finalizat în loc de la început.
Păstrează fiecare segment idempotent. Reprocesarea unui segment după o reîncercare nu trebuie să ducă la scrieri duble — leagă scrierile de înregistrarea sau ID-ul extern pe care îl procesezi.
Preferă un lanț segmentat în locul unei dispersări uriașe atunci când munca lovește o terță parte cu limită de rată: un lanț cu delayMs își dozează singur ritmul, în timp ce mii de joburi puse în coadă deodată devin toate eligibile imediat.

Dezvoltatori

Aplicații

API

Autogazda

Contribuie

Biblioteca UI

Prezentare generală

Pune rulările în coadă

Interogați periodic joburile

Alege propriul ID al lucrării

Opțiuni job

Reîncercați după un eșec temporar

Folosește-l: parcurge pe pagini o sincronizare lungă

Dispersare pe înregistrare

Bune practici pentru muncă de lungă durată

​Prezentare generală

​Pune rulările în coadă

​Interogați periodic joburile

​Alege propriul ID al lucrării

​Opțiuni job

​Reîncercați după un eșec temporar

​Folosește-l: parcurge pe pagini o sincronizare lungă

​Dispersare pe înregistrare

​Bune practici pentru muncă de lungă durată

Prezentare generală

Pune rulările în coadă

Interogați periodic joburile

Alege propriul ID al lucrării

Opțiuni job

Reîncercați după un eșec temporar

Folosește-l: parcurge pe pagini o sincronizare lungă

Dispersare pe înregistrare

Bune practici pentru muncă de lungă durată