Прогресс и наблюдение за задачей
HTTP-постановка возвращает jobId, но человеку нужен ответ о дальнейшей работе: задача ждёт, выполняется, завершена или отказала. Добавим чтение сохранённого состояния и сообщения прогресса. В «Импортёре» прогресс будет описывать две фазы: данные проверены и пакет применён. Это точнее случайного процента, который не учитывает очереди ожидания, retries и стоимость разных участков.
Состояние stages/11 содержит GET /api/imports/:id и отдельный наблюдатель QueueEvents. Сервер, worker и события не запускались, статус not_run. Ожидаемые объекты ниже показывают договор, а не результаты наблюдения реальной очереди. API остаётся локальным с вымышленными данными; проверка владельца задачи для production здесь не реализована и должна быть добавлена отдельно.
Прогресс является сообщением processor
Worker вызывает updateProgress после проверки CSV, перед applyShipment, и после его успешного возврата. Первая фаза сообщает, что массив пригоден для обработки. Вторая сообщает, что малый пакет принят источником и его result доступен. Если после второй фазы worker потеряет связь до completed, задача ещё может восстановиться через повтор. Поэтому committed и completed обозначают связанные, но разные этапы двух систем.
Следующие фрагменты находятся внутри processor в worker.js. Rows получены из проверенного файла, connection принадлежит worker, а applyShipment импортирован из apply.js.
await job.updateProgress({ phase: 'validated', rows: rows.length });
const result = await applyShipment(connection, batchId, rows);
await job.updateProgress({ phase: 'committed', rows: rows.length });
return result;Rows означает число позиций, а не количество единиц товара. Для этого малого пакета оно равно трём. Мы не имитируем постепенный прогресс по каждой строке, потому что окончательный Lua применяет весь пакет как одну операцию. Показать 33, 66 и 100 процентов во время предварительного цикла было бы неверно: человек мог бы считать часть остатков уже сохранённой, хотя запись ещё не началась.
Сохранённое состояние читается отдельно
Маршрут получает задачу через queue.getJob, затем читает getState, progress, returnvalue и failedReason. При отсутствующей записи возвращает 404. В этом состоянии retention ещё не ограничен, однако следующий урок объяснит, что после удаления такой ответ не доказывает отсутствие прежнего импорта. На первом этапе удобнее показать каждое поле отдельно, чтобы не скрывать его смысл за одной неопределённой надписью «готово».
Ожидаемый ответ после успешного выполнения может выглядеть так. Значения приведены как схема договора, не извлечены из запущенного Redis.
{
"id": "shipment-shipment-a-2026",
"state": "completed",
"progress": {"phase":"committed","rows":3},
"result": {"imported":3,"batchId":"shipment-a-2026"},
"error": null
}Запрос состоит из нескольких чтений, поэтому это не атомарный снимок всего жизненного цикла. Задача может изменить состояние между getJob и getState. Для пользовательского интерфейса небольшой временный переход обычно приемлем, если следующее чтение обновит его. Если нужен строгий отчёт о результате, следует определить отдельный устойчивый ресурс импорта или согласованную точку записи статуса. Простое перечисление полей не создаёт транзакционную гарантию.
QueueEvents наблюдает всю очередь
Локальный worker completed относится к одному процессу. QueueEvents получает события разных исполнителей той же очереди. Он использует общий queueName и prefix, а подключение работает в фоновом режиме. Полный events.js показывает отдельный наблюдатель: он печатает progress, completed и failed, а при остановке закрывает ресурс и соединение. Команда npm run events предназначена читателю и не выполнялась при подготовке.
import { QueueEvents } from 'bullmq';
import { connect, queueName, prefix } from './config.js';
const connection = connect('worker');
const events = new QueueEvents(queueName, { connection, prefix });
events.on('error', error => console.error('queue_events_error', error.message));
events.on('progress', ({ jobId, data }) => console.log('progress', jobId, data));
events.on('completed', ({ jobId }) => console.log('completed', jobId));
events.on('failed', ({ jobId, failedReason }) => console.error('failed', jobId, failedReason));
let closing = false;
async function shutdown() {
if (closing) return;
closing = true;
await events.close();
await connection.quit();
}
process.on('SIGTERM', () => shutdown().catch(console.error));
process.on('SIGINT', () => shutdown().catch(console.error));Событие completed содержит jobId, поэтому наблюдатель может связать его с запросом или перечитать сохранённый результат. Не следует использовать последнее событие независимо от ID: конкурентные задачи заканчиваются в разном порядке. События — сигнал о том, что состояние изменилось, а сохранённая задача или бизнес-ресурс — источник текущего результата. Такой договор помогает восстановить интерфейс после отключения наблюдателя без предположения, что он видел весь поток.
QueueEvents основан на Redis Streams, а не на обычном Pub/Sub. Однако его поток тоже имеет политику хранения и не должен считаться бессрочным аудитом всех импортов. Если наблюдатель отсутствовал дольше доступного окна, часть старых событий может быть удалена. Для долговременной отчётности нужен отдельный журнал с требованиями к сроку и полноте. Событие очереди и бизнес-документ имеют разные обязанности даже при похожих идентификаторах.
Повтор может вернуть раннюю фазу
При retry processor снова начинает проверку и может записать validated после ранее виденного committed. Это не означает, что остатки откатились: сохранённая done-отметка вернёт прежний result. Поэтому прогресс описывает текущую попытку, а не обязательно монотонную шкалу всего логического действия. В production полезно включить номер попытки или отдельный устойчивый статус бизнес-импорта, чтобы человек не считал повтор фазы потерей результата.
Если обновление progress завершилось ошибкой до записи, задача повторится без изменения. Если ошибка возникла после applyShipment, retry вернёт прежний result без новой прибавки. Прогресс является дополнительной операцией Redis и сам способен влиять на техническую попытку. Поэтому его нельзя считать бесплатным декоративным действием, которое никогда не ошибается. Идемпотентность основного эффекта делает эти неопределённые переходы управляемыми.
Не отправляйте событие progress на каждую строку большого CSV без необходимости. Это увеличивает количество команд и объём потока, хотя пользователь обычно воспринимает редкие значимые обновления. Полезно ограничить частоту или сообщать этапы, связанные с понятным результатом. При этом окончательное завершение требует отдельного явного события или сохранённого статуса, а не вывода, что последнее увиденное значение достаточно похоже на сто процентов.
Endpoint проверяет форму jobId, но не владельца. В локальном примере это допустимо только потому, что все товары вымышлены и адрес loopback. В настоящем сервисе наличие угаданного ID не должно разрешать чтение ошибки или returnvalue другого пользователя. Producer обязан связывать задачу с доверенным владельцем, а GET проверять эту связь перед выдачей данных. Не ставьте секрет в URL вместо такой серверной авторизации.
Прогресс и наблюдение описаны в BullMQ Events и Getters. После одиннадцатого состояния результат доступен по устойчивой ссылке и имеет объяснимые фазы. Последний шаг задаст срок хранения задач, завершение ресурсов и договор, по которому пример можно передать на сопровождение без выдуманных гарантий выпуска.