BatSchool · Очередь, которая помнит

* У старого моста через Анкмор было двенадцать сторожей и ни одного начальника. Когда сторож заболевал, его фонарь зажигал сосед. Город много раз хотел назначить им начальника, и каждый раз выяснялось, что мост светит, пока жив хоть один сторож, а начальник нужен только для отчёта.

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

Если в памяти башни, то до первого перезапуска. Башня встала, и всё, что она держала в голове, пропало: три тысячи писем снова лежат в мешках, и никто не знает, какие из них уже ушли. Задание должно лежать не в голове воркера, а там, где переживёт его смерть.

Очередь в голове башни

Очередь из урока 68 это Queue: файберы кладут в неё и берут из неё, и всё это в памяти процесса.

import { Effect, Queue, Schema } from 'effect';

const Job = Schema.Struct({ letter: Schema.Int, to: Schema.String });
type Job = typeof Job.Type;

export const tower = Effect.gen(function* () {
  const inbox = yield* Queue.unbounded<Job>();
  yield* Queue.offer(inbox, { letter: 7, to: 'T-3' });
  const job = yield* Queue.take(inbox);
  yield* Effect.log(`письмо ${job.letter} на ${job.to}`);
});

Для файберов одного процесса это хорошая очередь. Для линии башен плохая сразу по трём причинам:

Перезапуск процесса стирает её целиком.
Взятое задание исчезает из очереди сразу, ещё до того, как его сделали: упал на середине, и задания нет нигде.
Её видит только один процесс: вторая башня взять из неё не может.

Линии нужна сохраняемая очередь.

PersistedQueue

В Effect она называется PersistedQueue и живёт в модуле effect/persistence:

import { Effect, Layer, Schema } from 'effect';
import { PersistedQueue } from 'effect/persistence';

const Job = Schema.Struct({ letter: Schema.Int, to: Schema.String });

const night = Effect.gen(function* () {
  const jobs = yield* PersistedQueue.make({ name: 'liniya/jobs', schema: Job });
  yield* jobs.offer({ letter: 7, to: 'T-3' });
  yield* jobs.take((job, meta) => Effect.log(`письмо ${job.letter} на ${job.to}, попытка ${meta.attempts}`));
});

const Store = PersistedQueue.layer.pipe(Layer.provide(PersistedQueue.layerStoreMemory));

Effect.runPromise(night.pipe(Effect.provide(Store)));

[08:28:49.459] INFO (#1): письмо 7 на T-3, попытка 1

Разберём, что тут нового по сравнению с Queue.

У очереди есть имя, liniya/jobs. Две башни, которые открыли очередь с одним именем на одном хранилище, видят одни и те же задания. Имя это адрес очереди в хранилище.
У очереди есть схема. Задание уходит в хранилище закодированным и возвращается разобранным, как кадр Магистрали в уроке 76. Хранилище не знает, что такое Job: оно хранит закодированное значение в виде JSON.
take принимает не «дай задание», а обработчик. Задание считается сделанным, когда обработчик закончил успехом. Упал обработчик, и задание не пропало: очередь вернёт его.
Обработчик получает не только задание, но и meta.attempts: номер попытки, с единицы.
Где лежат задания, решает слой. PersistedQueue.layer это фабрика очередей, PersistedQueue.layerStoreMemory это хранилище в памяти. Для теста и для этого урока хватит памяти, на настоящей линии хранилище другое (об этом ниже), а код очереди тот же.

take с обработчиком это главная разница. У Queue.take ответственность кончается в момент выдачи: взял, и дальше сам. У PersistedQueue.take она кончается в момент успеха: пока обработчик не сказал «готово», задание принадлежит очереди.

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

Что значит «очередь вернёт задание»? Когда обработчик упал, задание прячется на паузу и потом снова становится видимым, уже со следующим номером попытки. Паузу задаёт расписание из урока 72, предел задаёт maxAttempts:

import { Effect, Schedule, Schema } from 'effect';
import { PersistedQueue } from 'effect/persistence';

export const Job = Schema.Struct({ letter: Schema.Int, to: Schema.String });

export const makeJobs = PersistedQueue.make({
  name: 'liniya/jobs',
  schema: Job,
  maxAttempts: 3,
  retrySchedule: Schedule.spaced('1 second'),
});

export const carryOne = Effect.gen(function* () {
  const jobs = yield* makeJobs;
  yield* jobs.take((job, meta) =>
    meta.attempts < 3 ? Effect.fail(`башня ${job.to} молчит`) : Effect.log(`письмо ${job.letter} ушло`),
  );
});

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

Без своих настроек очередь даёт заданию десять попыток, а пауза растёт от секунды и упирается в пять минут. Для писем это разумно: башня, которая молчит пять минут, вряд ли ответит через пять секунд, и стучать в неё каждую секунду значит стучать молотком, как сказала бы Тикка.

Предел нужен не меньше, чем повтор. Бывают задания, которые не выполнятся никогда: адрес, которого нет, письмо, которое не разбирается. Без предела такое задание крутится вечно и каждый раз съедает попытку воркера, которую он мог потратить на нормальное письмо.

Свой id против дублей

Липвиг носит письма мешками, и один мешок может прийти дважды: курьер не был уверен, что первый донёс. Если каждое offer это новое задание, письмо уйдёт дважды. Поэтому заданию можно дать свой id:

export const twice = Effect.gen(function* () {
  const jobs = yield* makeJobs;
  const first = yield* jobs.offer({ letter: 7, to: 'T-3' }, { id: 'letter-7' });
  const second = yield* jobs.offer({ letter: 7, to: 'T-3' }, { id: 'letter-7' });
  return [first, second];
});

Оба offer отдадут letter-7, а в очереди окажется одно задание. id это обещание: «задание с этим именем одно, сколько бы раз его ни положили». Номер письма для этого подходит лучше всего: у каждого письма он свой.

Задание на потом

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

export const later = Effect.flatMap(makeJobs, (jobs) =>
  jobs.offer({ letter: 9, to: 'T-5' }, { id: 'letter-9', delay: '1 hour' }),
);

Задание лежит в очереди, но час его не выдают. Задержка идёт по Clock, поэтому тест двигает TestClock и не ждёт настоящий час, как в уроке 74.

Покрути очередь сам: в виджете можно уронить обработчик посреди задания, поменять maxAttempts и паузу и посмотреть, как задание возвращается, переходит к другому воркеру и в конце концов снимается. Смерть самого воркера виджет не показывает: её закрывает аренда, о ней ниже.

Хранилище это слой

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

import { Layer } from 'effect';
import { PersistedQueue } from 'effect/persistence';
import { PgliteClient } from '@effect/sql-pglite';

export const InMemory = PersistedQueue.layer.pipe(Layer.provide(PersistedQueue.layerStoreMemory));

export const InBooks = PersistedQueue.layer.pipe(
  Layer.provide(PersistedQueue.layerStoreSql({ lockExpiration: '30 seconds' })),
  Layer.provide(PgliteClient.layer()),
);

layerStoreSql кладёт задания в таблицу той базы, которую даёт SqlClient из урока 94, а layerStoreRedis в Redis. У этих хранилищ есть то, чего нет у памяти: аренда. Взятое задание закреплено за воркером на lockExpiration, и воркер продлевает аренду раз в lockRefreshInterval, пока работает. Воркер умер, продлевать некому, и через lockExpiration задание снова видно другой башне. Это и есть «сосед зажигает фонарь»: никто не назначает, кому подхватить задание, оно само становится видно тем, кто жив.

Под капотом

Пондер Стиб · единственный, кто понимает Гекс

Внутри у каждого задания есть состояние и время, с которого его видно. Положили, и оно ждёт, видно сразу или после delay. Взяли, и оно занято. Обработчик закончил успехом, задание помечено выполненным и больше не выдаётся. Упал, и попытка засчитана: если попытки кончились, задание помечено проваленным, иначе снова ждёт, а время видимости сдвигается на паузу из расписания. Только прерванная работа попытку возвращает: остановить воркер штатно не значит потратить попытку. Расписание очередь проигрывает по номеру попытки, поэтому пауза растёт правильно, даже если вторую попытку делала другая башня, а не та, что первую.

Где тут дыра

declare const send: (job: { letter: number; to: string }) => Promise<void>;

const pending: Array<{ letter: number; to: string }> = [];

const work = async () => {
  for (;;) {
    const job = pending.shift();
    if (job === undefined) return;
    await send(job);
  }
};

На его примере всё работает: башня не падает, массив пустеет, письма уходят. Теперь соседний случай. send упал на письме 7, и письма 7 больше нет нигде: shift уже вынул его из массива. Башню перезапустили, и массив пуст, хотя писем было сорок.

А если Ринцвальд вставит try и будет класть упавшее письмо обратно в конец, письмо с адресом, которого нет, будет ходить по кругу до утра и отнимать каждую вторую попытку у настоящих писем. Задание должно принадлежать хранилищу, пока его не сделали, и иметь предел попыток. Оба свойства у PersistedQueue есть сразу.

Нестабильно. Модуль effect/persistence, включая PersistedQueue и его хранилища, в 4.1 помечен @stability unstable: имена и настройки могут поменяться в минорной версии. Урок написан на 4.1.0.

Линия, этап 1

Линия башен начинается с очереди заданий. Новый модуль src/queue.ts: задание Job, очередь liniya/jobs с тремя попытками и паузой в секунду, постановка по номеру письма, отложенная постановка и один шаг воркера. Тесты двигают часы, роняют обработчик и кладут одно письмо дважды.

EF-106-1★

В очереди с maxAttempts: 3 лежат письма 1, 2 и 3. Обработчик всегда падает на письме 2, остальные проходят. Воркер делает take снова и снова, двигая часы на секунду после каждой беды. Какие письма и с какими номерами попыток он увидит, пока очередь не опустеет? Что будет со следующим take?

EF-106-2★★

Липвиг предлагает класть письма без id: «дубли отсеет башня-получатель». Где появится дубль, если курьер положит мешок дважды, и что придётся помнить башне-получателю, чтобы его отсеять? Почему номер письма в id дешевле?

Практика

Глубже в архиве

Очереди, повторы и мёртвые письма без Effect разобраны в уроке про паттерны обмена сообщениями раздела про бэкенд.

Дальше

К утру мешки пустеют: задания помнит очередь, и перезапуск башни их больше не стирает. Пондер открывает журнал башни T-1 и долго считает.

Как писать в одну книгу без замков, расскажет следующий урок.