Фоновая обработка файлов

Одноразовый обработчик умеет создавать WebP, но после остановки может оставить processing навсегда. Добавим отдельный worker с SQL-очередью, временным владением задачей и ограниченными повторами. Задачами остаются карточки queued. API подтверждает версию, а worker забирает работу независимо от HTTP-запроса. Пользователь получает 202 быстро по контракту, хотя фактическая готовность зависит от обработки.

Главный механизм главы — lease, временное право фиксировать результат задачи. Он отличается от долгой блокировки транзакции: строку блокируем только для выбора и выдачи lease, затем COMMIT освобождает соединение. Декодирование и S3 PUT происходят за пределами транзакции. После работы условный UPDATE проверяет, что право фиксации всё ещё принадлежит текущей попытке.

Выбор задачи без общей очереди в памяти

Worker выполняет SELECT ... FOR UPDATE SKIP LOCKED LIMIT 1. Другой процесс пропустит уже заблокированную строку и сможет выбрать следующую. Такое поведение удобно для очереди, хотя не подходит для обычного пользовательского чтения, где пропущенная запись выглядела бы неожиданно исчезнувшей. Очередь сознательно допускает обработку следующей задачи без ожидания занятой первой.

Полный файл worker.mjs состояния 09:

import { randomUUID } from 'node:crypto';
import { setTimeout as pause } from 'node:timers/promises';
import { db,transaction } from './db.mjs';
import { derive } from './derive.mjs';
import { requireVersioning } from './store.mjs';
await requireVersioning(); let stopping=false;
process.on('SIGTERM',()=>{stopping=true;});
async function claim() {
  return transaction(async client=>{
    const q=await client.query(`SELECT * FROM archive_files WHERE
      (status='queued' AND available_at<=now()) OR
      (status='processing' AND lease_until<now())
      ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1`);
    if(!q.rowCount) return null; const token=randomUUID();
    const claimed=await client.query(`UPDATE archive_files SET status='processing',lease_token=$2,
      lease_until=now()+interval '2 minutes',attempts=attempts+1 WHERE id=$1 RETURNING *`,[q.rows[0].id,token]);
    return claimed.rows[0];
  });
}
while(!stopping) {
  const row=await claim(); if(!row){await pause(1000);continue;}
  try {
    const result=await derive(row);
    await db.query(`UPDATE archive_files SET status='ready',derived_key=$3,derived_version=$4,
      sha256=$5,lease_token=NULL,lease_until=NULL WHERE id=$1 AND lease_token=$2
      AND status='processing'`,[row.id,row.lease_token,result.key,result.version,result.sha256]);
  } catch {
    await db.query(`UPDATE archive_files SET status=CASE WHEN attempts>=3 THEN 'rejected' ELSE 'queued' END,
      error_code='processing_failed',available_at=now()+interval '30 seconds',lease_token=NULL,
      lease_until=NULL WHERE id=$1 AND lease_token=$2 AND status='processing'`,[row.id,row.lease_token]);
  }
}
await db.end();

Выбор и выдача lease находятся в одной транзакции. Worker создаёт случайный lease_token, устанавливает двухминутный lease_until и увеличивает attempts. Затем получает обновлённую строку. Если процесс завершится сразу после COMMIT, другой worker позднее увидит истёкший processing и заберёт задачу заново. Если процесс остановится до COMMIT, карточка останется queued и доступной сразу.

Время SQL now() используется для условий очереди и срока lease. Это уменьшает зависимость выбора от часов разных worker-машин. В остальной системе сроки подписей всё равно требуют нормальной синхронизации часов приложения и провайдера. Нельзя считать время базы универсальным исправлением всех расхождений часов: подпись S3 формируется другим компонентом и проверяется по его правилам.

Зачем нужен fencing token

Представим две попытки. Worker A получил lease, затем сильно задержался. Срок истёк, worker B получил новую попытку и новый token. A всё же завершил декодирование. Если он безусловно установит ready, то перезапишет состояние B или воскресит удаляемую карточку. Поэтому UPDATE требует одновременно UUID, свой lease_token и status='processing'.

Случайный token работает как fencing на уровне принятия результата. Он не останавливает старый процесс физически и не мешает ему потратить CPU либо записать лишний объект. Однако старый процесс уже не может сделать результат официальным в PostgreSQL. Производная имеет уникальный ключ, поэтому попытки не перезаписывают байты друг друга. Неиспользованный объект позднее попадёт в отложенную очистку.

В учебной версии lease не продлевается автоматически. Следовательно, внешняя граница времени обработки должна быть меньше двух минут либо нужно добавить heartbeat с тем же token. Иначе тяжёлое, но корректное изображение может обрабатываться повторно, пока предыдущая попытка ещё работает. Предел пикселей и размер помогают, однако не гарантируют конкретного времени на любой машине. Это эксплуатационное условие надо задать явно при собственном запуске.

Повторы и отказ

Ошибка обработки возвращает карточку в queued после тридцати секунд, а третья ошибка делает её rejected. В текущем примере безопасный код причины общий: processing_failed. Такой контракт не раскрывает путь, подписанный URL или содержимое исключения. Для промышленной диагностики можно выделить ограниченный набор причин: неподдерживаемое содержимое, временный отказ хранилища, превышение ресурсного лимита. Не стоит превращать произвольный текст исключения в публичный код.

Наш счётчик увеличивается при выдаче каждой попытки, включая восстановление истёкшего lease. При повторяющихся авариях процесса карточка может снова забираться после третьего claim, поскольку авария не выполняет обработчик отказа. Это граница простого worker: для строгого числа попыток требуется отдельный переход exhausted при claim или надзор над crashed попытками. Считать показанный счётчик гарантией ровно трёх запусков было бы неправильно.

Повтор полезен для временной недоступности S3, но не исправляет повреждённый PNG. Поэтому в большом сервисе ошибки содержимого делают окончательными сразу, а технические ошибки повторяют с увеличивающимся интервалом и случайным разбросом. Урок оставляет короткую фиксированную задержку, чтобы было видно устройство процесса. Массовый одновременный повтор без jitter способен ещё сильнее нагрузить уже отказавшую зависимость.

Важна наблюдаемость самой очереди. Количество queued, возраст старейшей готовой к выбору задачи, число processing с истёкшим lease и доля rejected описывают разные проблемы. Большая очередь при работающих worker означает нехватку производительности; много истёкших lease указывает на остановки или слишком короткий срок. Список пользовательских файлов не заменяет эти эксплуатационные показатели.

Завершение процесса

На SIGTERM worker перестаёт брать новые задачи и старается завершить текущую. После выхода из цикла закрывается пул PostgreSQL. Если инфраструктура оборвёт процесс раньше завершения, механизм истечения lease даст следующему worker возможность повторить работу. Для реального выпуска время graceful shutdown согласуют с максимальным временем pipeline. Нельзя обещать завершение при любом принудительном останове.

С девятого состояния process-one.mjs больше не входит в снимок: используется worker, управляющий теми же карточками. Выберите полную папку lesson-09, чтобы не смешать прежний ручной обработчик с lease-протоколом. В материалах не измерены скорость очереди, повторяемость ошибок или предел памяти; исходники только подготовлены и прочитаны редакционно. Эти параметры требуют отдельной лабораторной проверки.

Ожидаемая последовательность для допустимой фотографии теперь имеет вид pending → queued → processing → ready. Вторая попытка не меняет официальную производную, если её lease уже потерян. Следующий урок применит тот же принцип к удалению: смена состояния отзовёт право фиксации, а отдельный проход завершит очистку байтов после нужного защитного интервала.

Защита результата проверяется также по числу изменённых строк. Нулевой UPDATE после успешного PUT означает потерю права фиксации, а не успешное принятие. В расширенной реализации worker записывает безопасное событие lease_lost и передаёт UUID производной сборке сирот. Сам объект при этом не выдаётся пользователю, потому что карточка его не зарегистрировала.

Предыдущий урок: Безопасные производные изображения · Следующий урок: Удаление файла и гонки операций

Оглавление серии · Исходники всех состояний.

Документация: PostgreSQL: SKIP LOCKED, node-postgres: транзакции.