Job in background

Panoramica

L’esecuzione di una logic function è limitata dal suo timeoutSeconds (massimo 900 secondi). Qualsiasi attività che non può essere completata in tale intervallo — una risincronizzazione completa, un fan-out per record, un’API di terze parti che applica limitazioni di frequenza — deve essere suddivisa in esecuzioni più piccole. enqueueJobs fa esattamente questo: chiede ai worker di Twenty di eseguire in un secondo momento una delle logic function della tua app, una volta per payload, ogni esecuzione nel proprio processo con il proprio budget di timeout. Il chiamante restituisce immediatamente.

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

Mettere in coda esecuzioni

Importa enqueueJobs da twenty-sdk/logic-function, puntalo all’universalIdentifier della logic function da eseguire e passa un payload per ogni esecuzione da mettere in coda.

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 } }],
});

Ogni esecuzione riceve il proprio payload come argomento dell’handler, esattamente come qualsiasi altro trigger. La destinazione deve appartenere alla stessa applicazione del chiamante — l’inserimento in coda della funzione di un’altra app viene rifiutato con Logic function not found e non viene inserito nulla in coda. Una singola chiamata accetta fino a 200 job.

enqueueJobs restituisce non appena i job sono stati accettati, non quando sono stati eseguiti. Non restituisce i risultati delle destinazioni: fai scrivere a ciascuna destinazione ciò che produce nel key-value store o in un record dello spazio di lavoro se hai bisogno di leggerlo di nuovo. Per scoprire se un’esecuzione è terminata, esegui il polling del relativo stato del job.

Il precedente helper enqueueJob, che mette in coda un singolo job per chiamata, è deprecato. Usa invece enqueueJobs con un elenco jobs di un solo elemento. Anche l’argomento payloads è deprecato: passa invece jobs, che accetta lo stesso payload per ogni esecuzione e ti consente di assegnare un nome a ciascuna.

Eseguire il polling dei job

enqueueJobs restituisce un jobId per ogni esecuzione accodata, nello stesso ordine dei job passati. getJobs legge a che punto sono arrivate tali esecuzioni, con una sola chiamata.

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

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

const statuses = await getJobs(jobIds);

Una singola chiamata a getJobs legge al massimo 200 ID, lo stesso limite che enqueueJobs applica a un batch. Ogni voce contiene l’ID richiesto:

Campo	Cosa contiene
`jobId`	L’ID usato per il polling.
`stato`	`WAITING`, `PRIORITIZED`, `DELAYED`, `ACTIVE`, `WAITING_CHILDREN`, `COMPLETED` o `FAILED`.
`attemptsMade`	Tentativi della coda effettuati finora, incluso quello corrente.
`failedReason`	Il messaggio di errore dell’ultimo tentativo non riuscito, altrimenti `null`.
`enqueuedAt`	Quando il job è stato accettato, in millisecondi epoch.
`startedAt`	Quando un worker lo ha prelevato, `null` mentre è ancora in attesa.
`finishedAt`	Quando è stato completato o non è riuscito, `null` mentre è ancora in esecuzione.

Un job che non è ancora stato prelevato viene normalmente indicato come PRIORITIZED anziché WAITING, perché i job dell’applicazione vengono accodati con una priorità bassa fissa. COMPLETED e FAILED sono gli stati terminali. Un ciclo di polling dovrebbe anche interrompersi autonomamente dopo un numero limitato di tentativi anziché attendere per sempre, poiché un’esecuzione è leggibile solo finché la coda la conserva: 4 ore dopo il completamento, 7 giorni dopo un errore e, in entrambi i casi, solo gli ultimi 1000 job in tale stato. getJobs restituisce solo i job che conserva ancora, quindi un ID che è stato rimosso — o che non è mai esistito — è semplicemente assente dal risultato anziché generare un errore per l’intero batch. Confronta ciò che ottieni con ciò che hai richiesto per individuarli e considera un ID mancante come “non più noto”, che non equivale a “non è mai stato eseguito”. Gli stati sono limitati all’area di lavoro da cui viene effettuata la chiamata, quindi un’area di lavoro non può mai leggere il job di un’altra area di lavoro.

Scegli il tuo ID job

Passa jobs invece di payloads per assegnare un nome a ogni esecuzione. L’ID che passi è quello accettato da getJobs, quindi un chiamante può effettuare il polling di un job senza dover memorizzare ciò che ha restituito enqueueJobs.

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

Un ID è composto da un massimo di 128 caratteri tra lettere, cifre, _, . o -, e nessun job in una chiamata può condividerlo con un altro. Passa payloads oppure jobs, mai entrambi. Quando ometti jobId, Twenty ne genera uno. Un jobId è anche una chiave di idempotenza: l’accodamento di un ID che la coda conserva ancora viene accettato, ma non avvia una seconda esecuzione, e getJobs continua a segnalare la prima. Ciò rende sicuro ritentare l’accodamento. Due avvertenze: la garanzia dura solo finché il job viene conservato, quindi lo stesso ID può avviare una nuova esecuzione dopo che l’originale è stato rimosso, e gli ID sono condivisi da tutte le app in un’area di lavoro, quindi aggiungi un prefisso ai tuoi se una collisione sarebbe rilevante.

Opzioni del job

Le opzioni si applicano a ogni esecuzione nel batch.

Opzione	Predefinito	Intervallo	Cosa fa
`retryLimit`	`0`	`0`–`10`	Tentativi aggiuntivi complessivi nella coda. I nuovi tentativi richiesti dall’applicazione sono limitati a `3`. Aumenta questo valore solo per gli handler che possono essere eseguiti due volte in sicurezza.
`delayMs`	`0`	`0`–`604800000` (7 giorni)	Attendi questo intervallo prima che le esecuzioni diventino idonee.

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

La priorità non è ancora configurabile. I job in coda vengono sempre eseguiti alla priorità più bassa, così il lavoro della piattaforma non viene mai ritardato rispetto ai job dell’applicazione. Il controllo sulla priorità sarà disponibile a breve.

L’esecuzione in coda eredita l’utente attivo della funzione che l’ha messa in coda, quindi agisce con le stesse autorizzazioni.

Riprova un errore transitorio

Twenty non ritenta ogni eccezione del codice applicativo. Un normale errore generato viene trattato come un errore permanente. In caso di errore transitorio, una funzione logica in coda può richiedere fino a tre tentativi generando 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}`,
    );
  }
};

Genera direttamente RetryableLogicFunctionError quando possibile. Se la estendi, non sostituire il relativo name: Twenty riconosce il nome serializzato RetryableLogicFunctionError in tutti i runtime di esecuzione. retryCount è 0 per l’esecuzione iniziale e aumenta solo quando il codice applicativo richiede un nuovo tentativo. maxRetries è al massimo 3 e può essere inferiore quando il job in coda ha un limite complessivo di tentativi inferiore. Gli errori della piattaforma non aumentano retryCount, sebbene consumino comunque il budget di sicurezza complessivo della coda. La coda ritarda i tentativi con backoff esponenziale e jitter. Il ritardo esatto non è intenzionalmente garantito, quindi il codice applicativo non deve dipendere dal fatto che un nuovo tentativo avvenga in un momento preciso. Una volta raggiunto maxRetries, un ulteriore RetryableLogicFunctionError viene registrato come errore finale dell’applicazione senza un’altra esecuzione.

I nuovi tentativi rieseguono l’intero gestore e possono verificarsi dopo che alcuni effetti collaterali sono riusciti. Rendi idempotente il gestore prima di richiedere nuovi tentativi.

Usalo per scorrere una lunga sincronizzazione a pagine

La forma classica è una funzione che mette in coda se stessa con il cursore successivo. Ogni esecuzione gestisce una pagina di lavoro ben all’interno del proprio timeout, e la catena si interrompe quando non resta più nulla.

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,
});

Suddivisione per record

Quando il lavoro è naturalmente per elemento, metti in coda un job per elemento in una singola chiamata e lascia che i worker li elaborino in parallelo invece di ciclare in linea.

const companies = await listCompaniesToEnrich();

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

Buone pratiche per il lavoro di lunga durata

Due regole coprono quasi ogni job lungo: usa la ricorsione invece del ciclo e elabora un blocco limitato per esecuzione. Un’esecuzione che prova a fare tutto è una modalità di errore: raggiunge il timeout e, con un nuovo tentativo, ricomincia tutto da zero. Invece, dimensiona un blocco in modo che finisca comodamente entro timeoutSeconds, conserva la tua posizione e metti in coda l’esecuzione successiva.

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,
});

Cosa rende tutto questo solido:

Dimensiona il blocco a partire dall’elemento più lento, non dalla media. CHUNK_SIZE × worst-case item time deve rientrare in timeoutSeconds con un certo margine, altrimenti la parte finale di un blocco va persa quando l’esecuzione viene interrotta.
Rendi esplicita la condizione di terminazione. Usa la ricorsione solo finché torna un blocco completo. Una catena che si interrompe solo su “nessun risultato” continuerà all’infinito se la fonte restituisce mai una pagina corta a metà percorso.
Conserva i progressi prima di mettere in coda l’esecuzione successiva, così un anello fallito riparte dall’ultimo blocco completato invece che dall’inizio.
Mantieni ogni blocco idempotente. Rielaborare un blocco dopo un ritentativo non deve causare scritture duplicate: usa chiavi di scrittura sul record o sull’id esterno che stai elaborando.
Preferisci una catena a blocchi a un’unica enorme suddivisione parallela quando il lavoro colpisce una terza parte con limitazione di velocità: una catena con delayMs si autoregola, mentre migliaia di job messi in coda in una volta sola diventano tutti idonei immediatamente.

Sviluppatori

App

API

Auto-ospitazione

Contribuire

Libreria UI

Panoramica

Mettere in coda esecuzioni

Eseguire il polling dei job

Scegli il tuo ID job

Opzioni del job

Riprova un errore transitorio

Usalo per scorrere una lunga sincronizzazione a pagine

Suddivisione per record

Buone pratiche per il lavoro di lunga durata

​Panoramica

​Mettere in coda esecuzioni

​Eseguire il polling dei job

​Scegli il tuo ID job

​Opzioni del job

​Riprova un errore transitorio

​Usalo per scorrere una lunga sincronizzazione a pagine

​Suddivisione per record

​Buone pratiche per il lavoro di lunga durata

Panoramica

Mettere in coda esecuzioni

Eseguire il polling dei job

Scegli il tuo ID job

Opzioni del job

Riprova un errore transitorio

Usalo per scorrere una lunga sincronizzazione a pagine

Suddivisione per record

Buone pratiche per il lavoro di lunga durata