Очереди и фоновые задачи: когда пора выносить работу из запроса

Симптом всегда один: страница «думает» и падает по таймауту. Внутри — генерация отчёта, отправка сотни писем или выгрузка в стороннюю систему. Пользователь ждёт, браузер обрывает соединение, работа прерывается на середине, а данные остаются в непонятном состоянии. Лечится это не увеличением таймаута, а переносом работы в фон.

Что обязательно выносить из запроса

Минимальная очередь

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

create table jobs (
  id           bigserial primary key,
  kind         text not null,
  payload      jsonb not null,
  run_at       timestamptz not null default now(),
  attempts     int not null default 0,
  max_attempts int not null default 5,
  locked_at    timestamptz,
  finished_at  timestamptz,
  last_error   text
);

create index jobs_pending on jobs (run_at)
  where finished_at is null;

-- skip locked позволяет запустить несколько обработчиков:
-- каждый берёт свои задачи и они не конфликтуют
update jobs set locked_at = now(), attempts = attempts + 1
where id = (
  select id from jobs
  where finished_at is null and locked_at is null and run_at <= now()
  order by run_at
  limit 1
  for update skip locked
)
returning *;

Ключевая конструкция здесь — `for update skip locked`. Она позволяет держать несколько обработчиков одновременно: каждый забирает свою задачу, и две копии одной работы не запускаются.

Повторы и предел попыток

Задачи падают, это нормально. Важно, чтобы падение не превращалось ни в бесконечный цикл, ни в тихую потерю. Схема, которая работает: экспоненциальная задержка, ограниченное число попыток, отдельная таблица окончательно проваленных задач с возможностью перезапуска руками.

async function runJob(job: Job) {
  try {
    await handlers[job.kind](job.payload)
    await db.update(jobs).set({ finishedAt: new Date() }).where(eq(jobs.id, job.id))
  } catch (error) {
    const failedForGood = job.attempts >= job.maxAttempts

    if (failedForGood) {
      // Не удаляем и не прячем: задача должна попасть человеку на разбор
      await moveToDeadLetter(job, error)
      reportMetric("jobs.dead_letter", 1, { kind: job.kind })
      return
    }

    // 1 мин, 2, 4, 8... — даём внешней системе шанс восстановиться
    const delayMinutes = 2 ** job.attempts
    await db.update(jobs)
      .set({ lockedAt: null, runAt: minutesFromNow(delayMinutes), lastError: String(error) })
      .where(eq(jobs.id, job.id))
  }
}

За чем следить

• Длина очереди: устойчивый рост означает, что обработчики не успевают
• Возраст самой старой незавершённой задачи — точнее длины показывает задержку
• Число окончательно проваленных задач по типам
• Время выполнения по типам: рост обычно предшествует отказу

Из этих четырёх показателей самый полезный — возраст старейшей задачи. Длина очереди может быть большой при нормальной работе, а вот задача, которая висит час, почти всегда означает проблему.

Очередь не ускоряет работу. Она делает срок выполнения независимым от терпения пользователя и от доступности чужих систем.