Постановка импорта в BullMQ

Оглавление серии

После проверки частоты запрос может создать работу, которую не нужно выполнять внутри HTTP-ответа. Добавим очередь BullMQ. Producer — часть приложения, ставящая задачу, — проверит контракт и вернёт идентификатор. Worker, отдельный исполнитель, появится следующим уроком. Ответ 202 будет означать, что задача поставлена, а не что остатки уже изменены и импорт завершён.

Снимок stages/05 использует BullMQ 6.3.12 и ioredis 5.11.1, версии подтверждены по release-tag package.json. Redis остаётся отдельным локальным экземпляром с политикой noeviction. Пакеты, очередь и сервер не запускались, статус not_run. Все ответы являются ожидаемыми. У серии нет YouTube-источника: договор и вымышленные данные созданы для самостоятельного урока по первичной документации.

Задача содержит ссылку на известный набор

В HTTP-теле передаются dataset и batchId. Dataset может иметь только значение shipment-a, которое соответствует файлу из снимка. Не принимаем произвольный путь к файлу: producer не должен превращать пользовательскую строку в доступ к любому каталогу машины. BatchId обозначает логический пакет поступления. Он остаётся одинаковым при повторе одного действия и меняется, если действительно нужно применить другое поступление.

Данные задачи также содержат contractVersion: 1. Это версия предметного договора между producer и worker, а не версия BullMQ. Воркер позднее проверит её независимо от HTTP-обработчика. Такое дублирование нужно, потому что задача может остаться в Redis после обновления HTTP-кода, а другой producer способен поставить объект без нашего маршрута. Проверка на входе очереди не отменяет проверку перед изменением основных данных.

Полный файл queue.js создаёт очередь с явным именем и префиксом. Соединение передаётся из конфигурации, поэтому здесь нет скрытого нового адреса Redis.

import { Queue } from 'bullmq';
import { queueName, prefix } from './config.js';

export function createQueue(connection, options = {}) {
  const queue = new Queue(queueName, { connection, prefix, ...options });
  queue.on('error', error => console.error('queue_error', error.message));
  return queue;
}

Имя importer-local различает очередь задачи, а prefix importer-demo отделяет её ключи. Producer и worker должны использовать оба значения одинаково. Если различается имя или префикс, постановка может успешно завершиться, но исполнитель будет читать другую очередь. Это выглядит как «воркер ничего не делает», хотя причина не в обработке CSV. Конфигурация принадлежит общему договору участников, как название события в чате.

JobId подавляет повтор постановки

При добавлении задаём jobId: shipment-{batchId}. В данной очереди одинаковый ID не создаёт ещё одну сохранённую задачу, пока прежняя запись существует. Используем дефис вместо двоеточия, которое имеет специальное значение в именах ключей BullMQ. Ограничиваем batchId алфавитом букв, цифр и дефисов, а также длиной до шестидесяти четырёх знаков. ID остаётся публичным указателем результата, не секретом и не доказательством владельца.

Фрагмент маршрута POST /api/imports из server.js выполняется после rate limit и проверки тела. queue создан показанной функцией, а request и response предоставлены Express.

const job = await queue.add('import-shipment', {
  contractVersion: 1,
  dataset,
  batchId
}, {
  jobId: `shipment-${batchId}`
});
response.status(202).json({ jobId: job.id });

Ожидаемое тело ответа для batchId shipment-a-2026 содержит идентификатор shipment-shipment-a-2026. Повтор того же HTTP-действия получает ссылку на ту же задачу, если она сохранена. Пока воркер отсутствует, задача остаётся ожидающей. Нельзя возвращать число импортированных строк из producer: он ещё не читал и не применял CSV, а успешное добавление в очередь не подтверждает пригодность данных для изменения остатков.

{"jobId":"shipment-shipment-a-2026"}

Постановка и бизнес-эффект различаются

Уникальность jobId действует в области очереди и срока существования её записи. Если завершённая задача позже удалена политикой retention, повторная постановка того же ID может снова стать допустимой. Даже без новой постановки сама задача может выполняться повторно после ошибки или потери блокировки. Поэтому jobId не обеспечивает однократное прибавление остатков. Восьмой урок добавит отдельную бизнес-отметку пакета вместе с изменением количества.

В локальном примере все запросы относятся к одному общему набору данных. Production-система должна включать владельца или организацию в область batchId и проверять право читать задачу. Иначе чужой человек сможет угадывать идентификатор, получить результат или занять ID, который ожидал другой клиент. Префикс очереди изолирует имена технически, но не является границей авторизации HTTP-ресурса.

Если Redis недоступен, queue.add() может завершиться ошибкой, и producer вернёт 503. В конфигурации HTTP-подключение имеет ограниченное количество повторов и не сохраняет команды в offline-очереди. Это помогает ответу не ждать бесконечно восстановления зависимости. При этом возможен неопределённый результат сетевого разрыва: Redis успел добавить задачу, но HTTP-клиент не получил 202. Повтор с тем же batchId помогает найти прежнюю постановку.

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

Очередь не является Pub/Sub

Pub/Sub передаёт сообщение текущим подписчикам, но отключённый подписчик его пропускает. Для фонового импорта требуется сохранённое состояние задачи, ожидание исполнителя и политика повторов. Поэтому используем BullMQ, а не выдаём публикацию события в Redis за долговечную очередь. В дальнейшем QueueEvents также имеет собственный механизм событий, но источник бизнес-результата должен оставаться в сохранённой задаче и основных данных.

Политика noeviction важна для ключей очереди. Если Redis удаляет произвольные ключи ради освобождения памяти, структура сохранённой задачи может нарушиться. TTL кеша допустим, потому что копия восстанавливается; ключи очереди и отметки импорта имеют другие обязанности. При нехватке памяти приложение должно получить объяснимую ошибку и остановить постановку, а не надеяться, что вытеснение выберет только ненужные cache-ключи.

Конструктор и соединения описаны в BullMQ Connections, область ID — в Job Ids, а граница Pub/Sub — в Redis Pub/Sub. После пятого состояния допустимый запрос имеет сохранённую задачу и устойчивую ссылку. Дальше отдельный процесс прочитает CSV и изменит остатки, оставив HTTP-обработчик свободным для других запросов.