Hintergrundjobs

Ein Logikfunktionslauf ist durch seine timeoutSeconds begrenzt (maximal 900 Sekunden). Alles, was in diesem Zeitfenster nicht abgeschlossen werden kann – ein vollständiger Re-Sync, ein Fan-out pro Datensatz, eine Drittanbieter-API, die dich rate-limitiert – muss in kleinere Läufe aufgeteilt werden. enqueueJobs macht genau das: Es weist die Twenty-Worker an, eine der Logikfunktionen deiner App später, einmal pro Payload, jeweils in einem eigenen Prozess mit eigenem Timeout-Budget auszuführen. Der Aufrufer kehrt sofort zurück.

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

Läufe in die Warteschlange stellen

Importiere enqueueJobs aus twenty-sdk/logic-function, gib den universalIdentifier der auszuführenden Logikfunktion an und übergib zum Einreihen eine Payload pro Lauf.

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

Jeder Lauf erhält seine Payload als Handler-Argument, genau wie jeder andere Trigger. Das Ziel muss zur gleichen Applikation gehören wie der Aufrufer — das Einreihen der Funktion einer anderen App wird mit Logic function not found abgelehnt und nichts wird eingereiht. Ein einzelner Aufruf akzeptiert bis zu 200 Payloads.

enqueueJobs kehrt zurück, sobald die Jobs akzeptiert wurden, nicht erst, wenn sie ausgeführt wurden. Sie gibt nicht die Ergebnisse der Ziele zurück — lasse jedes Ziel das, was es erzeugt, in den Key-Value-Speicher oder in einen Workspace-Datensatz schreiben, wenn du es wieder auslesen musst.

Der ältere Helper enqueueJob, der pro Aufruf einen einzelnen Job einreiht, ist veraltet. Verwende stattdessen enqueueJobs mit einer payloads-Liste mit einem Element.

Job-Optionen

Optionen gelten für jeden Lauf im Batch.

Option	Standard	Bereich	Was es tut
`retryLimit`	`0`	`0`–`10`	Zusätzliche Warteschlangenversuche insgesamt. Von der Anwendung angeforderte Wiederholungsversuche sind auf `3` begrenzt. Erhöhe diesen Wert nur für Handler, die sicher zweimal ausgeführt werden können.
`delayMs`	`0`	`0`–`604800000` (7 Tage)	So lange warten, bevor die Läufe zur Ausführung anstehen.

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

Priorität ist noch nicht konfigurierbar. Eingereihte Jobs laufen immer mit der niedrigsten Priorität, sodass Plattformarbeit niemals hinter Applikationsjobs verzögert wird. Steuerung der Priorität kommt bald.

Der eingereihte Lauf erbt den agierenden Benutzer der Funktion, die ihn eingereiht hat, sodass er mit denselben Berechtigungen agiert.

Wiederholen eines vorübergehenden Fehlers

Twenty wiederholt nicht jede Ausnahme aus dem Anwendungscode. Ein gewöhnlicher ausgelöster Fehler wird als dauerhafter Fehler behandelt. Bei einem vorübergehenden Fehler kann eine eingereihte Logikfunktion durch Auslösen von RetryableLogicFunctionError bis zu drei Wiederholungen anfordern.

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

Löse RetryableLogicFunctionError nach Möglichkeit direkt aus. Wenn du sie erweiterst, ersetze ihren name nicht: Twenty erkennt den serialisierten Namen RetryableLogicFunctionError in allen Laufzeitumgebungen. retryCount ist bei der initialen Ausführung 0 und erhöht sich nur, wenn der Anwendungscode eine Wiederholung anfordert. maxRetries beträgt höchstens 3 und kann geringer sein, wenn der Job in der Warteschlange ein niedrigeres Gesamtlimit für Wiederholungen hat. Plattformfehler erhöhen retryCount nicht, verbrauchen jedoch weiterhin das Gesamtsicherheitsbudget der Warteschlange. Die Warteschlange verzögert Wiederholungsversuche mit exponentiellem Backoff und Jitter. Die genaue Verzögerung wird absichtlich nicht garantiert, daher sollte der Anwendungscode nicht davon abhängen, dass eine Wiederholung zu einem genauen Zeitpunkt erfolgt. Sobald maxRetries erreicht ist, wird ein weiterer RetryableLogicFunctionError ohne weitere Ausführung als endgültiger Anwendungsfehler erfasst.

Wiederholungen führen den gesamten Handler erneut aus und können erfolgen, nachdem einige Nebenwirkungen erfolgreich waren. Mache den Handler idempotent, bevor du Wiederholungen anforderst.

Verwendung: eine lange Synchronisation seitenweise verarbeiten

Die klassische Form ist eine Funktion, die sich selbst mit dem nächsten Cursor in die Warteschlange stellt. Jeder Lauf erledigt eine Seite Arbeit deutlich innerhalb seines eigenen Timeouts, und die Kette endet, wenn nichts mehr übrig ist.

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

Aufteilung pro Datensatz

Wenn die Arbeit natürlich pro Element anfällt, reihe in einem einzigen Aufruf einen Job pro Element ein und lasse die Worker sie parallel verarbeiten, anstatt inline zu iterieren.

const companies = await listCompaniesToEnrich();

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

Bewährte Vorgehensweisen für langlaufende Arbeit

Zwei Regeln decken fast jeden langen Job ab: rekursiv statt in einer Schleife arbeiten und pro Lauf einen begrenzten Chunk verarbeiten. Ein Lauf, der versucht, alles zu erledigen, ist der Fehlerfall — er erreicht das Timeout, und bei einem Retry startet er den gesamten Vorgang erneut von vorn. Bestimme stattdessen die Größe eines Chunks so, dass er bequem innerhalb von timeoutSeconds fertig wird, speichere deine Position und reihe den nächsten Lauf ein.

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

Warum das funktioniert:

Lege die Chunk-Größe anhand des langsamsten Elements fest, nicht anhand des Durchschnitts. CHUNK_SIZE × Worst-Case-Elementzeit muss mit ausreichendem Puffer in timeoutSeconds passen, sonst geht das Ende eines Chunks verloren, wenn der Lauf abgeschnitten wird.
Mache die Abbruchbedingung explizit. Rekursiere nur, solange ein voller Chunk zurückgekommen ist. Eine Kette, die nur bei „keine Ergebnisse“ anhält, läuft für immer weiter, wenn die Quelle unterwegs einmal eine verkürzte Seite zurückgibt.
Persistiere den Fortschritt, bevor du den nächsten Lauf einreihst, damit ein fehlgeschlagener Link beim letzten abgeschlossenen Chunk und nicht am Anfang neu startet.
Halte jeden Chunk idempotent. Die erneute Verarbeitung eines Chunks nach einem Retry darf nicht zu doppelten Schreibvorgängen führen — führe Schlüsselschreibvorgänge auf dem Datensatz oder der externen ID aus, die du verarbeitest.
Bevorzuge eine verkettete Chunk-Verarbeitung gegenüber einem riesigen Fan-out, wenn die Arbeit auf einen rate-limitierten Drittanbieter trifft: Eine Kette mit delayMs taktet sich selbst, wohingegen Tausende auf einmal eingereihte Jobs sofort zur Ausführung anstehen.

Übersicht

Apps

API

Selbsthosting

Mitwirken

Läufe in die Warteschlange stellen

Job-Optionen

Wiederholen eines vorübergehenden Fehlers

Verwendung: eine lange Synchronisation seitenweise verarbeiten

Aufteilung pro Datensatz

Bewährte Vorgehensweisen für langlaufende Arbeit

​Läufe in die Warteschlange stellen

​Job-Optionen

​Wiederholen eines vorübergehenden Fehlers

​Verwendung: eine lange Synchronisation seitenweise verarbeiten

​Aufteilung pro Datensatz

​Bewährte Vorgehensweisen für langlaufende Arbeit

Läufe in die Warteschlange stellen

Job-Optionen

Wiederholen eines vorübergehenden Fehlers

Verwendung: eine lange Synchronisation seitenweise verarbeiten

Aufteilung pro Datensatz

Bewährte Vorgehensweisen für langlaufende Arbeit