Job in background

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',
  payloads: [{ 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 payload.

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.

Il precedente helper enqueueJob, che mette in coda un singolo job per chiamata, è deprecato. Usa invece enqueueJobs con un elenco payloads di un solo elemento.

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',
  payloads: [{ 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,
      payloads: [{ 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,
  payloads: companies.map((company) => ({ 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,
      payloads: [{ 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.

Panoramica

App

API

Auto-ospitazione

Contribuire

Mettere in coda esecuzioni

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

​Mettere in coda esecuzioni

​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

Mettere in coda esecuzioni

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