Зависшая задача и потеря блокировки

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

Задача может перестать продвигаться, даже если processor не выбросил обычную ошибку. Worker способен завершиться, потерять связь или надолго занять event loop, из-за чего не обновит блокировку активной задачи. BullMQ обнаруживает такую ситуацию как stalled и допускает повторное выполнение. Рассмотрим, почему потеря технической блокировки не означает отмену уже применённого поступления.

Состояние stages/09 использует идемпотентный applyShipment из предыдущего урока. Код, остановки процессов и потеря блокировки не запускались, статус not_run. Параметры времени выбраны для понятного договора локального примера, а не для демонстрации измеренного сбоя. Никакие тяжёлые циклы, нагрузочные сценарии или команды управления Redis при подготовке не выполнялись.

Блокировка обозначает владение работой

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

В worker.js девятого снимка явно указаны lockDuration, stalledInterval и maxStalledCount. Ниже полный файл, включая процессор и подписку на событие stalled.

import { Worker, UnrecoverableError } from 'bullmq';
import { connect, queueName, prefix } from './config.js';
import { readCatalog, productsKey, generationKey } from './catalog.js';
import { readShipment } from './csv.js';
import { applyShipment } from './apply.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);
  }
  const result = await applyShipment(connection, batchId, rows);

  return result;
}, {
  connection,
  prefix,
  lockDuration: 30000,
  stalledInterval: 30000,
  maxStalledCount: 1,
  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));
worker.on('stalled', jobId => console.error('job_stalled', jobId));

LockDuration задаёт период действия блокировки, а stalledInterval относится к проверке застрявшей работы. Не нужно складывать их и обещать точный момент повтора в интерфейсе: реальное обнаружение зависит от циклов проверки, состояния Redis и работающих исполнителей. Событие stalled помогает диагностировать ситуацию, но не является таймером с гарантией до миллисекунды. После возврата в waiting задача ещё ждёт доступного worker.

MaxStalledCount равен единице. Это ограничивает число разрешённых восстановлений после потери владения, а не обычные attempts после выброшенного Error. Оба механизма могут привести к повторному вызову processor, поэтому безопасность изменения должна быть одинаковой независимо от причины. Если защищать запись только счётчиком attemptsMade, другая причина повторного выполнения останется вне выбранного условия.

Event loop участвует в продлении

Node.js выполняет обычный JavaScript в event loop. Долгий синхронный расчёт способен задержать таймеры и обработку сетевых ответов, даже когда процесс не завершился. Поэтому worker может выглядеть живым для операционной системы и всё же потерять возможность своевременно продлевать блокировку. Высокое значение concurrency не исправляет блокирующий цикл: несколько задач того же процесса разделяют тот же поток обычного JavaScript.

Наш CSV ограничен сотней строк, чтение файла асинхронно, а разбор мал. Это снижает стоимость учебной операции, но не является измерением её максимальной длительности. Если добавить большие вычисления или сложную синхронную обработку, их нужно вынести в подходящие отдельные процессы либо разбивать так, чтобы управление возвращалось event loop. Простое увеличение lockDuration скрывает некоторые симптомы, но не устраняет причину задержки.

Lua также требует внимания: Redis выполняет наш малый пакет как одну операцию, временно не обслуживая другие команды. Большой скрипт может задерживать обновления блокировок остальных задач, даже если Node.js свободен. Поэтому предел размера пакета относится не только к памяти, но и к общему времени владения Redis. Нельзя объявлять любую синхронную атомарную запись бесплатной для всех участников очереди.

Повтор после уже выполненного изменения

Рассмотрим ожидаемую последовательность: worker получил задачу A, применил остатки и записал done, но остановился до перехода BullMQ в completed. Очередь позже обнаруживает потерянную блокировку и выдаёт A другому worker. Новый исполнитель прочитает тот же CSV, вызовет applyShipment и получит сохранённый result. Остатки не увеличатся повторно, хотя processor действительно запускался два раза. Это модель поведения, не фактический аварийный вывод.

Worker 1: пакет A применён, done сохранён
Worker 1 исчез до completed
stalled → waiting
Worker 2: тот же A → прежний result, новые остатки не создаются

Если первый worker исчез до применения, done отсутствует и второй выполняет пакет впервые. Сетевой разрыв посередине ответа Lua создаёт неопределённость у Node.js, однако сохранённая отметка после успешно выполненного скрипта всё равно направляет повтор к прежнему результату. Нам не нужно угадывать момент сбоя по консольному сообщению; предметный договор содержит собственную точку принятия действия.

С технической блокировкой не следует строить бизнес-уникальность. При потере lock прежний исполнитель иногда ещё способен продолжать внешнюю операцию, а другой уже получил задачу. Для платежа или отправки письма нужна идемпотентность внешнего API либо согласованный журнал эффекта. В нашем примере весь предметный эффект помещается в Redis-скрипт, поэтому повтор контролируется там, где изменяются остатки, а не только там, где worker владеет job.

Сопровождение видит несколько состояний

Событие stalled содержит jobId и позволяет связать потерю блокировки с задачей. Событие failed после исчерпания границы указывает окончательный технический отказ. Но основные данные могли уже измениться раньше, поэтому статус failed не равен отсутствию бизнес-эффекта. Для ответа человеку нужно проверить result и done по batchId или спроектировать единый ресурс импорта, который объединяет эти сведения в понятный статус.

Если stalled появляется часто, сначала изучают нагрузку event loop, время Redis-команд, длительность внешних операций и число процессов. Нельзя автоматически увеличивать все интервалы до часов: это заставит реально исчезнувшую задачу долго ждать восстановления. Настройка времени всегда балансирует терпимость к временной задержке и скорость обнаружения потерянного исполнителя. Выбранное значение должно следовать из наблюдений в собственном окружении, которых этот курс не заявляет.

Принудительное завершение и аккуратная остановка различаются. Позднее worker.close() перестанет получать новую работу и дождётся активных задач, сохраняя возможность нормального завершения. При аварии это ожидание невозможно, поэтому нужен механизм stalled. Обе процедуры должны сохранять один бизнес-договор повтора; нельзя писать обработчик, который корректен только когда человек заранее предупреждает процесс о завершении.

Современная серия не добавляет устаревший QueueScheduler ради этих задач. Выбранная версия BullMQ имеет собственную актуальную модель worker и проверок; API старых примеров необходимо сверять с документацией текущей версии. Случайное создание лишнего класса из исторического урока не улучшает идемпотентность и может вообще не соответствовать установленному пакету. В снимке используются только прочитанные актуальные API.

Потеря владения описана в BullMQ Stalled Jobs. После девятого состояния причина возможного повтора не скрыта за общим словом «ошибка». Следующим шагом добавим несколько одновременно выполняемых задач и рассмотрим, какие гарантии остаются при конкуренции разных пакетов.