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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

const companies = await listCompaniesToEnrich();

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

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

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

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

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

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

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

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

Обзор

Приложения

API

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

Внести вклад

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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