Воркер и обработка CSV

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

Producer уже возвращает идентификатор поставленной задачи, но очередь сама не читает CSV и не меняет товары. Добавим worker — отдельный процесс, который получает задачу import-shipment, проверяет её договор, читает разрешённый файл и прибавляет количество к текущим остаткам. HTTP-сервер продолжает отвечать независимо от времени обработки. Разделение особенно важно, когда импорт перестаёт быть мгновенной операцией.

Шестой снимок stages/06 вводит намеренно простой неидемпотентный путь записи. Он показывает, что делает исполнитель, и создаёт исходный случай для уроков о повторе и атомарности. Используйте только вымышленные данные отдельного локального Redis. Пример not_run: worker, CSV-разбор и изменения остатков не выполнялись, ожидаемые количества ниже рассчитаны по учебному договору.

Исполнитель проверяет собственный вход

Воркер получает объект job от BullMQ. Проверяем имя задачи, версию договора, dataset и batchId. Мы не полагаемся только на HTTP-проверку: задача может остаться от другого выпуска или появиться из другого producer. До любого изменения основных данных требуется убедиться, что исполнителю понятен её смысл. Неподдерживаемая версия не должна молча превращаться в новый импорт по частично совпавшим полям.

Подключение worker использует maxRetriesPerRequest: null, поскольку фоновый процесс может ждать восстановления Redis. Producer имеет иной режим для ограниченного времени HTTP-ответа. Это не противоречие настроек, а различие задач двух процессов. Создавать один общий клиент с бесконечным ожиданием для HTTP и считать его подходящим всем участникам очереди было бы неудобно человеку, ожидающему ответ формы.

Полный файл worker.js шестого состояния показывает обработчик, запись и локальные события исполнителя. Команда читателя npm run worker открывает отдельный процесс и при подготовке не выполнялась.

import { Worker, UnrecoverableError } from 'bullmq';
import { connect, queueName, prefix } from './config.js';
import { readCatalog, productsKey, generationKey } from './catalog.js';
import { readShipment } from './csv.js';

const connection = connect('worker');
const worker = new Worker(queueName, async job => {
  if (job.name !== 'import-shipment') throw new UnrecoverableError('unknown_job');
  const { dataset, batchId, contractVersion } = job.data;
  if (contractVersion !== 1 || dataset !== 'shipment-a' ||
      typeof batchId !== 'string' || !/^[a-zA-Z0-9-]{1,64}$/.test(batchId)) {
    throw new UnrecoverableError('invalid_contract');
  }
  await readCatalog(connection);
  let rows;
  try {
    rows = await readShipment(dataset);
  } catch (error) {
    throw error;
  }

  for (const row of rows) {
    const raw = await connection.hget(productsKey, row.sku);
    const old = raw ? JSON.parse(raw) : { quantity: 0 };
    const product = { ...row, quantity: old.quantity + row.quantity };
    await connection.hset(productsKey, row.sku, JSON.stringify(product));
  }
  await connection.incr(generationKey);
  const result = { imported: rows.length, batchId };

  return result;
}, {
  connection,
  prefix,
  concurrency: 1
});

worker.on('error', error => console.error('worker_error', error.message));
worker.on('failed', (job, error) => console.error('job_failed', job?.id, error.message));
worker.on('completed', job => console.log('job_completed', job.id));

Функция readCatalog() сначала гарантирует начальное заполнение отсутствующих полей. Затем основной источник уже находится в hash, а обработка прибавляет значения CSV к существующим остаткам. Повторное начальное заполнение не заменяет остатки, потому что использует HSETNX. Это важно при двух задачах одного процесса: каждый вызов должен читать текущее количество, а не возвращать начальные значения fixture перед применением нового пакета.

Ограниченный CSV-диалект

Файл shipment-a.csv содержит заголовок sku,title,quantity и три строки. Каждая строка описывает количество поступления, а не окончательный остаток. Используем простой учебный диалект без кавычек, запятых внутри названия и многострочных полей. Такой разбор нельзя выдавать за универсальный CSV-парсер; формат ограничен договором заранее. Для реального загружаемого файла нужен подходящий парсер и отдельные пределы размера, числа строк и времени обработки.

Полный csv.js сначала читает только разрешённый dataset, проверяет заголовок и длину, затем возвращает проверенный массив. Повтор SKU внутри пакета отклоняется, чтобы смысл количества не зависел от случайного двойного появления одной позиции.

import { readFile } from 'node:fs/promises';

export async function readShipment(dataset) {
  if (dataset !== 'shipment-a') throw new Error('unknown_dataset');
  const text = await readFile(new URL('./fixtures/shipment-a.csv', import.meta.url), 'utf8');
  const lines = text.trim().split(/\r?\n/);
  if (lines.shift() !== 'sku,title,quantity') throw new Error('invalid_header');
  if (lines.length > 100) throw new Error('too_many_rows');
  const seen = new Set();
  return lines.map((line, index) => {
    const fields = line.split(',');
    const [sku, title, rawQuantity] = fields;
    const quantity = Number(rawQuantity);
    if (fields.length !== 3 || !/^p-[1-9][0-9]*$/.test(sku) ||
        !title || title.length > 60 || !Number.isSafeInteger(quantity) ||
        quantity < 1 || quantity > 1000 || seen.has(sku)) {
      throw new Error(`invalid_row_${index + 2}`);
    }
    seen.add(sku);
    return { sku, title, quantity };
  });
}

Данные проверяются до записи первой строки. Это позволяет отклонить неправильное количество третьего товара без уже выполненного изменения первых двух. Однако после проверки цикл записи всё ещё состоит из отдельных команд Redis и может оборваться на середине из-за сетевого сбоя. Предварительная проверка сокращает одну группу частичных результатов, но не превращает несколько изменений в атомарный пакет. Эту оставшуюся проблему важно видеть отдельно.

Ожидаемый результат поступления

Начальный остаток лампы равен 8, блокнота — 3. CSV добавляет соответственно 12 и 5, а новый карандаш получает 4. После успешного выполнения ожидается лампа 20, блокнот 8 и карандаш 4. Воркер возвращает imported 3 и batchId. BullMQ сохраняет результат задачи, а каталог обновляет generation, чтобы следующие чтения использовали новый cache-ключ. Это ожидаемые данные, не результат запущенного импорта.

p-1 Лампа: 8 + 12 = 20
p-2 Блокнот: 3 + 5 = 8
p-3 Карандаш: 0 + 4 = 4
result.imported = 3

Число imported обозначает количество строк, а не количество единиц товара. В пакете всего 21 единица, но три позиции. Похожие числовые поля легко смешать в интерфейсе и отчёте, поэтому их смысл задаётся предметным договором. Если понадобится сумма quantity, её нужно вернуть отдельным явно названным значением, не меняя незаметно значение imported для уже существующего клиента.

Локальные события completed и failed относятся к этому worker. Если задача завершилась другим процессом, данный обработчик completed не обязан сработать. Для общего наблюдения позже добавим QueueEvents. Результат самого processor также не равен консольной строке: возвращённый объект становится частью сохранённого состояния задачи, а лог лишь помогает человеку сопровождать конкретный процесс.

Частичный результат и повтор

Если сеть пропала после прибавления лампы, но до остальных строк, источник уже частично изменён. BullMQ может позднее повторить задачу, и первая строка прибавится снова. Даже успешный полный цикл может повториться после потери подтверждения завершения или блокировки. Поэтому текущий путь специально нельзя считать безопасным для реального учёта остатков. Он позволяет увидеть, почему «очередь умеет повторы» недостаточно для правильного бизнес-результата.

READ старого JSON и HSET нового также образуют гонку между двумя воркерами. Оба могут прочитать остаток 8, прибавить разные количества и записать свои результаты, потеряв одну прибавку. Пока concurrency равен 1 в одном процессе, риск выглядит незаметным, однако другой worker уже создаёт параллельность. Следующие уроки должны устранить её атомарным изменением, а не надеяться, что приложение всегда будет состоять из одного исполнителя.

Воркер увеличивает generation только после цикла. Если процесс завершится раньше, основные данные могут измениться, а кеш останется старого поколения. Это ещё одна граница неатомарной записи. Восьмой урок объединит остатки, поколение и бизнес-отметку пакета в одном Lua-скрипте. Тогда успешный результат будет относиться ко всему малому пакету, а не к случайной части последовательных команд.

Создание исполнителя и различие соединений описаны в BullMQ Workers и Connections. После шестого состояния поток работы завершён от HTTP-постановки до результата. Следующий шаг введёт управляемый повтор и постоянную ошибку, сохраняя явное предупреждение о неидемпотентности текущего пути изменения остатков.