Повтор задачи и постоянная ошибка

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

Во время фоновой работы встречаются ошибки двух разных видов. Временный сбой зависимости может исчезнуть, и повтор имеет смысл. Неверный заголовок CSV не исправится сам через секунду, поэтому бесконечная обработка только тратит ресурсы. Настроим в «Импортёре» ограниченное число попыток с задержкой и обозначим постоянные ошибки через UnrecoverableError.

Седьмой снимок stages/07 ещё содержит неидемпотентный цикл изменения остатков из предыдущего урока. Повтор способен увеличить количество дважды, поэтому пример предназначен для анализа вымышленных локальных данных. Код и сбои не запускались, статус not_run. Окончательный безопасный путь малого пакета появится следующим шагом; политика retry и правильность бизнес-изменения должны рассматриваться вместе.

Attempts считает все попытки

В producer задаём defaultJobOptions: attempts 3 и exponential backoff с исходной задержкой 1000 миллисекунд. Три — общее максимальное количество обычных попыток, включая первую. Это не три повтора после начального выполнения. При двух неудачах третий запуск исчерпывает выбранную границу; дальнейшее действие требует отдельного решения, а не незаметного вечного цикла.

Фрагмент объявления очереди находится в server.js. Соединение redis уже создано из config, а createQueue импортирован из queue.js.

const queue = createQueue(redis, {
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 }
  }
});

Экспоненциальная задержка увеличивает интервалы следующих повторов. Она помогает не создавать непрерывный поток запросов к неисправной зависимости. Однако задача в delayed не выполняется «ровно в назначенную миллисекунду»: время означает, что повтор становится доступным, а реальное получение зависит от наличия исполнителя и очереди. Интерфейс не должен обещать точное время завершения только по параметру backoff.

Если backoff отсутствует, повтор может начаться без выбранной паузы. Для временно недоступной зависимости это способно усилить нагрузку. Слишком длинная пауза, наоборот, задерживает полезное восстановление после краткого сбоя. В учебном контракте исходная секунда делает механизм понятным; production-настройка требует знания стоимости операции, допустимого ожидания и поведения внешней системы. При массовом сбое полезен также случайный разброс повторов.

Постоянный отказ не ждёт следующего запуска

Worker проверяет имя задачи и договор до записи. Неподдерживаемая версия, неизвестный dataset или неправильный batchId вызывают UnrecoverableError. CSV тоже проверяется заранее; в этом локальном снимке ошибка его чтения и формы считается непригодной конфигурацией задачи. Такое решение относится к известным локальным fixtures. Для удалённого хранилища файлов сетевой сбой чтения пришлось бы отличать от неправильного содержания.

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

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 new UnrecoverableError(error.message);
  }

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

BullMQ ожидает выброшенный объект Error, а не строку или число. Тип ошибки нужен библиотеке для правильного учёта и сохранения причины. throw "bad" может выглядеть кратко, но создаёт неудобный и недостаточно определённый договор. Для ожидаемой постоянной ошибки используем специальный тип; для временной оставляем обычный Error с диагностическим кодом. Публичный API позднее должен показывать человеку подходящее объяснение, не раскрывая внутренний стек.

Ошибка задачи и ошибка процесса

Событие worker failed относится к конкретной задаче. Событие worker error сообщает проблему самого исполнителя или его подключения. Их нельзя объединять в одну надпись «CSV неверен»: потеря Redis не доказывает, что данные файла испорчены. В логах сохраняем идентификатор задачи рядом с failed, чтобы связать отказ с batchId и последующей диагностикой. Детальные данные пользователя или секрет соединения для этой связи не нужны.

Ожидаемая последовательность временного сбоя: первая попытка выбрасывает Error, задача получает задержку, следующая снова вызывает processor. После успешного выполнения возвращается обычный result и задача становится completed. Постоянная ошибка формата должна завершиться failed без дальнейших предусмотренных retries. Эти переходы описывают контракт библиотеки и снимка; они не являются наблюдением выполненной аварийной проверки.

Временный Error → задержка → следующая попытка
Успех следующей попытки → completed
UnrecoverableError → failed без автоматического retry

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

Повтор не является откатом

Неудача processor не возвращает основные данные к прежнему состоянию. Если первая строка уже прибавлена, а вторая завершилась ошибкой, автоматический повтор начнёт обработку всего массива снова. Ни состояние failed, ни attemptsMade не определяют, какие предметные изменения успели произойти. Они описывают жизненный цикл задачи. Для безопасного результата нужен идемпотентный обработчик или транзакционная запись, а не только аккуратная задержка.

Рассмотрим лампу: исходное 8 плюс 12 даёт 20. Если задача упала после этой записи и повторилась, новая попытка может получить 20 и добавить 12 ещё раз, получив 32. Ответ completed будет технически успешным, но бизнес-результат ошибочным. Такой пример показывает, почему тест «задача в итоге завершилась» не доказывает корректность импорта. Проверять следует остатки и связь с логическим пакетом, а не один статус очереди.

Количество попыток также не гарантирует отсутствие повторного выполнения по другим причинам. Потеря блокировки или остановка worker может перевести задачу из active обратно в waiting. Отдельный урок объяснит stalled и связь с event loop. Бизнес-обработчик должен допускать повтор независимо от того, пришёл он через обычный retry, восстановление исполнителя или ручное действие сопровождения. Нельзя защищать остатки только условием attemptsMade равным нулю.

Если ошибка произошла до любого изменения, повтор пока безвреден для количества. Но разработчик не всегда может доказать момент сбоя по отсутствующему ответу команды: Redis мог выполнить HSET и потерять связь раньше подтверждения. Поэтому безопасный договор строится вокруг сохранённого результата логического действия. Он должен выдерживать неопределённый сетевой исход, а не только очевидные ошибки до отправки запроса.

Attempts, backoff и постоянные ошибки описаны в Retrying failing jobs. После седьмого состояния retries ограничены и непригодные задачи прекращаются осмысленно. Следующий урок устранит повторное прибавление, объединяя основные остатки, смену поколения и отметку принятого пакета одной операцией Redis.