Хранение результатов и контракт выпуска

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

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

Последний снимок stages/12 полный, однако имеет статус not_run. Установка зависимостей, Redis, HTTP, worker, наблюдатель, тесты, сборка и деплой не выполнялись. Показанные состояния являются ожидаемыми результатами договора. Курс создаёт локальные артефакты и не подключается к реальной инфраструктуре. Архив содержит только исходники и вымышленные fixtures, без node_modules, секретов и случайных рабочих данных.

Срок задачи не равен сроку поступления

Для completed задаём age 3600 секунд и count 100, для failed — age 86400 и count 100. Политика ограничивает возраст и количество сохранённых задач. Уборка происходит при последующих переходах задач, поэтому это не отдельный точный таймер, который гарантированно удалит запись в заданную секунду. В интерфейсе следует понимать, что технический результат может исчезнуть, а не считать getJob вечной справкой о каждом поступлении.

Фрагмент объявления очереди в server.js показывает окончательные defaultJobOptions. CreateQueue импортирован из queue.js, а redis является producer-подключением.

const queue = createQueue(redis, {
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: { age: 3600, count: 100 },
    removeOnFail: { age: 86400, count: 100 }
  }
});

После удаления завершённой задачи тот же jobId может снова использоваться при постановке. Однако applyShipment найдёт бизнес-отметку done и не прибавит остатки повторно. Это и есть причина разделить уникальность очереди и идемпотентность предметного действия. Техническая уборка не должна молча разрешать второе поступление, если batchId по договору продолжает обозначать прежний неизменный пакет.

Done-отметки в учебном примере бессрочны. Это создаёт отдельный рост данных, который нельзя игнорировать только потому, что очередь очищается. Для production нужен долговечный журнал импортов и правило архивирования отметок. Если срок идемпотентности ограничен, пользователь и producer должны знать, когда старое действие снова может иметь эффект. Удаление всех done по аналогии с кешем меняет бухгалтерский смысл количества, а не просто освобождает копии.

Аккуратная остановка worker

Worker.close() прекращает получение новой работы и ждёт текущие задачи. Затем закрываем предоставленное подключение Redis через quit. Не закрываем соединение раньше ожидания, иначе processor потеряет зависимость во время записи и завершения. Если операция никогда не заканчивается, сама close тоже может ждать; production требует внешнего срока остановки и идемпотентного восстановления после принудительного завершения. В учебном малом пакете срок ограничен входным договором, но не измерен.

Фрагмент завершающей части worker.js использует ранее созданные worker и connection. Повторный сигнал не запускает процедуру второй раз.

let closing = false;
async function shutdown() {
  if (closing) return;
  closing = true;
  await worker.close();
  await connection.quit();
}
process.on('SIGTERM', () => shutdown().catch(console.error));
process.on('SIGINT', () => shutdown().catch(console.error));

HTTP-процесс отдельно прекращает приём соединений, затем закрывает Queue и producer-клиент. Наблюдатель отдельно закрывает QueueEvents. Эти ресурсы не исчезают потому, что завершён другой процесс: три участника имеют собственные соединения и собственные обработчики сигналов. Именно раздельная структура позволяет обновлять интерфейс или наблюдение независимо от выполнения, но требует явного сопровождения каждого жизненного цикла.

Готовность и диагностические данные

Маршрут health проверяет флаг closing и делает ping Redis. Он показывает, что процесс намерен обслуживать работу и видит зависимость. Он не подтверждает наличие свободного worker, пригодность CSV, место на диске для резервирования или успешность конкретной задачи. Для реального сервиса полезно отдельно наблюдать длину waiting, возраст старейшей задачи, число failed и stalled, время обработки и ошибки подключения. Один зелёный ping не описывает всю цепочку.

Ошибка failedReason хранится для диагностики, но production-API не должен бездумно раскрывать внутренние адреса, пути файлов и секретные параметры. Пользователю обычно нужен код с объяснением исправимого действия, а сопровождению — связь с jobId и batchId. Логи не должны копировать весь реальный CSV ради удобства. Договор наблюдения определяет собираемые поля, срок хранения и доступ к ним так же явно, как договор импорта определяет количество.

После retention GET задачи может вернуть 404, хотя пакет был успешно применён. Такой ответ означает отсутствие технической записи очереди, а не отсутствие бизнес-события. Если человеку нужна история поступлений, её следует читать из отдельного устойчивого ресурса, основанного на бизнес-журнале. В нашем примере done содержит минимальный result, но HTTP-маршрут долговременной истории специально не реализован и не обещается через getJob.

Версии producer и worker

ContractVersion 1 должен поддерживаться всеми активными исполнителями. Если новый producer начинает ставить версию 2 раньше обновления worker, прежний processor получит постоянный отказ. При выпуске выбирают совместимый переход: сначала исполнитель, понимающий обе версии, затем новый producer, затем удаление старой поддержки после обработки оставшихся задач. Конкретный порядок зависит от изменения договора; это план совместимости, а не выполненный деплой.

Версии пакетов зафиксированы, но lockfile не создан без установки зависимостей. При самостоятельном воспроизведении читатель получает его в своём окружении и сохраняет вместе с результатами проверки. BullMQ 6 имеет отличия от 5, включая удаление старых repeatable-job API; наша серия их не использует. Обновлять major-версию только заменой цифры в package.json без чтения миграционного договора неправильно даже для похожего примера.

Redis должен иметь noeviction, а долговечность зависит от persistence и резервирования. RDB, AOF и восстановление имеют разные свойства; очередь сама не гарантирует сохранение последней подтверждённой записи после потери хранилища. Основные продукты, generation и done следует восстанавливать согласованно. Если восстановить отметки без остатков или наоборот, повтор даст неправильный эффект. Конфигурация хранения входит в предметный контракт, хотя этот курс не выполняет её настройку или аварийное восстановление.

Политики описаны в Auto-removal of jobs, остановка — в Graceful shutdown, сохранность — в Redis persistence. Распакуйте архив «Импортёра» в отдельную папку, выберите один stages/NN и собственный учебный Redis. Полученный путь завершает договор от кеша до результата очереди, явно сохраняя границы хранения, повторов и выпуска.