Фоновые задания

Запуск логической функции ограничен значением timeoutSeconds (максимум 900 секунд). Любая задача, которая не может завершиться за это время — полная повторная синхронизация, разветвление по каждой записи, сторонний API, который ограничивает вас по скорости, — должна быть разделена на более мелкие запуски. enqueueJobs делает именно это: он просит воркеров Twenty позже запустить одну из логических функций вашего приложения, по одному разу для каждой полезной нагрузки, каждый запуск в отдельном процессе с собственным лимитом времени. Вызов возвращается немедленно.

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

Постановка запусков в очередь

Импортируйте enqueueJobs из twenty-sdk/logic-function, укажите universalIdentifier логической функции для запуска и передайте в очередь по одной полезной нагрузке для каждого запуска.

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

Каждый запуск получает свою полезную нагрузку в качестве аргумента обработчика, точно так же, как любой другой триггер. Целевая функция должна принадлежать тому же приложению, что и вызывающая сторона — постановка в очередь функции другого приложения будет отклонена с ошибкой Logic function not found, и ничего не будет поставлено в очередь. Один вызов принимает до 200 полезных нагрузок.

enqueueJobs возвращает управление, как только задания приняты, а не когда они выполнены. Он не возвращает результаты целевых функций — если вам нужно потом их прочитать, пусть каждая целевая функция запишет то, что она создаёт, в key-value store или в запись рабочего пространства.

Старый помощник enqueueJob, который добавляет в очередь одно задание за вызов, устарел. Вместо этого используйте enqueueJobs со списком payloads, состоящим из одного элемента.

Параметры задания

Параметры применяются к каждому запуску в пакете.

Вариант	По умолчанию	Диапазон	Что делает
`retryLimit`	`0`	`0`–`10`	Общее количество дополнительных попыток постановки в очередь. Количество повторных попыток, запрошенных приложением, ограничено значением `3`. Увеличивайте это значение только для обработчиков, которые безопасно запускать дважды.
`delayMs`	`0`	`0`–`604800000` (7 дней)	Сколько ждать, прежде чем запуски станут допустимыми к выполнению.

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

Приоритет пока нельзя настраивать. Задания в очереди всегда выполняются с самым низким приоритетом, поэтому работа платформы никогда не откладывается из‑за заданий приложений. Возможность управлять приоритетом появится скоро.

Запуск в очереди наследует действующего пользователя функции, которая его поставила в очередь, поэтому он работает с теми же правами.

Повторная попытка при временном сбое

Twenty не выполняет повторную попытку при каждом исключении из кода приложения. Обычная вызванная ошибка считается постоянным сбоем. При временном сбое логическая функция в очереди может запросить до трёх повторных попыток, вызвав 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}`,
    );
  }
};

По возможности вызывайте RetryableLogicFunctionError напрямую. Если вы расширяете его, не заменяйте его name: Twenty распознаёт сериализованное имя RetryableLogicFunctionError во всех средах выполнения. retryCount равен 0 при первоначальном выполнении и увеличивается, только когда код приложения запрашивает повторную попытку. maxRetries не превышает 3 и может быть меньше, если у задания в очереди меньше общий лимит повторных попыток. Сбои платформы не увеличивают retryCount, хотя всё равно расходуют общий резерв безопасности очереди. Очередь задерживает повторные попытки с экспоненциальной задержкой и случайным разбросом. Точная задержка намеренно не гарантируется, поэтому код приложения не должен зависеть от выполнения повторной попытки в точно определённое время. После достижения maxRetries ещё один RetryableLogicFunctionError фиксируется как окончательный сбой приложения без нового выполнения.

Повторные попытки заново выполняют весь обработчик и могут происходить после успешного выполнения некоторых побочных эффектов. Сделайте обработчик идемпотентным, прежде чем запрашивать повторные попытки.

Использование: постраничный проход по длинной синхронизации

Классический шаблон — функция, которая ставит саму себя в очередь со следующим курсором. Каждый запуск обрабатывает одну страницу работы, уверенно укладываясь в собственный таймаут, а цепочка останавливается, когда больше нечего делать.

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

Разветвление по каждой записи

Когда работа естественным образом выполняется по отдельным элементам, добавляйте в очередь по одному заданию на элемент за один вызов и позвольте воркерам обрабатывать их параллельно, вместо того чтобы перебирать элементы в цикле непосредственно в коде.

const companies = await listCompaniesToEnrich();

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

Рекомендации для длительной работы

Два правила покрывают почти любое долгое задание: рекурсивный вызов вместо цикла и обработка ограниченного блока за один запуск. Запуск, который пытается сделать всё сразу, — это сценарий отказа: он упирается в таймаут и при повторной попытке начинает всё сначала с нуля. Вместо этого выберите размер блока так, чтобы он с запасом завершался в пределах timeoutSeconds, сохраните свою позицию и поставьте в очередь следующий запуск.

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

Почему это работает:

Размер блока выбирайте, исходя из самого медленного элемента, а не среднего. CHUNK_SIZE × время обработки в худшем случае должно помещаться в timeoutSeconds с запасом, иначе хвост блока будет потерян, когда запуск обрежет таймаут.
Сделайте условие завершения явным. Вызывайте рекурсивно только пока вернулся полный блок. Цепочка, которая останавливается только на основе «нет результатов», будет продолжаться бесконечно, если источник когда‑либо вернёт укороченную страницу посередине.
Сохраняйте прогресс перед постановкой в очередь следующего запуска, чтобы при сбое звена цепочки она возобновлялась с последнего завершённого блока, а не с начала.
Сделайте каждый блок идемпотентным. Повторная обработка одного блока после повторной попытки не должна приводить к двойной записи — привязывайте операции записи к идентификатору обрабатываемой записи или к внешнему идентификатору.
Отдавайте предпочтение цепочке из блоков перед одним гигантским разветвлением, когда работа обращается к внешнему сервису с ограничением по скорости: цепочка с delayMs саморегулирует темп, тогда как тысячи заданий, поставленных в очередь одновременно, сразу становятся допустимыми к выполнению.

Обзор

Приложения

API

Самостоятельный хостинг

Внести вклад

Постановка запусков в очередь

Параметры задания

Повторная попытка при временном сбое

Использование: постраничный проход по длинной синхронизации

Разветвление по каждой записи

Рекомендации для длительной работы

​Постановка запусков в очередь

​Параметры задания

​Повторная попытка при временном сбое

​Использование: постраничный проход по длинной синхронизации

​Разветвление по каждой записи

​Рекомендации для длительной работы

Постановка запусков в очередь

Параметры задания

Повторная попытка при временном сбое

Использование: постраничный проход по длинной синхронизации

Разветвление по каждой записи

Рекомендации для длительной работы