BatSchool · Старая книга: постепенный переход на Effect

В уроке 117 мы строили город так, как хотели бы с самого начала. Настоящий город так не строится: почти любой проект, куда ты придёшь с Effect, уже живёт: в нём есть промисы, колбэки, сервер на Hono или Express и библиотеки, которые никто не будет переписывать.

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

Старый клиент

Вот что лежит в подвале. Трогать этот код нельзя:

// Архив Магистрали, клиент на промисах.
interface Letter {
  readonly id: string;
  readonly to: string;
}

interface LegacyArchive {
  find(id: string): Promise<Letter | undefined>;
  list(offset: number, limit: number): Promise<ReadonlyArray<Letter>>; // пустая страница значит конец
  subscribe(listener: (letter: Letter) => void): { unsubscribe(): void };
  end(): Promise<void>;
}

declare function openArchive(url: string): Promise<LegacyArchive>;

У клиента четыре привычки, и каждая знакома по прошлым урокам. Соединение надо открыть и не забыть закрыть: это ресурс из урока 55. Любой промис может отказать, и тип об этом молчит. Список отдаётся страницами. Новые письма приходят колбэком, от которого надо отписаться.

Одна дверь: паттерн `use`

Первая мысль: завернуть каждый метод в свой эффект. Через неделю методов станет двадцать, и у каждой обёртки будет свой способ переводить ошибки.

Лучше сделать одну дверь. Сервис отдаёт не методы старого клиента, а функцию use: дай мне, что сделать с клиентом, а я сделаю это и отвечу эффектом.

class ArchiveError extends Data.TaggedError('ArchiveError')<{ readonly message: string }> {}

const toArchiveError = (error: unknown) =>
  new ArchiveError({ message: error instanceof Error ? error.message : String(error) });

class Archive extends Context.Service<
  Archive,
  {
    readonly use: <A>(f: (archive: LegacyArchive) => Promise<A>) => Effect.Effect<A, ArchiveError>;
  }
>()('magistral/Archive') {}

const archiveLayer = (url: string) =>
  Layer.effect(
    Archive,
    Effect.gen(function* () {
      const archive = yield* Effect.acquireRelease(
        Effect.tryPromise({ try: () => openArchive(url), catch: toArchiveError }),
        (archive) => Effect.ignore(Effect.tryPromise(() => archive.end())),
      );
      return Archive.of({
        use: (f) => Effect.tryPromise({ try: () => f(archive), catch: toArchiveError }),
      });
    }),
  );

Три решения в двадцати строках.

Соединение живёт столько же, сколько слой. acquireRelease открывает архив, когда слой строят, и закрывает, когда слой разбирают. Это те же слои с ресурсом, что в уроке 56.
Каждый отказ промиса превращается в ArchiveError с текстом. Других путей наружу у старого кода нет, поэтому типизированная ошибка честна.
Отказ при закрытии Effect.ignore глотает: финализатор не должен ронять программу, которая уже уходит.

Код, который ищет письмо, теперь выглядит как любой другой код на Effect:

const findLetter = (id: string) =>
  Effect.flatMap(Archive, (archive) => archive.use((legacy) => legacy.find(id)));
// Effect<Letter | undefined, ArchiveError, Archive>

Новый метод старого клиента не требует новой обёртки: use уже умеет звать любой. А тесты подставят свой слой Archive, как в уроке 53.

Ошибок в типе нет, потому что Effect.promise обещает, что промис не откажет. Если откажет, это дефект, как в уроке 40: catchTag его не видит, окошко отвечает 500, а в типе написано never. Для чужих промисов всегда tryPromise: его отказ попадает в канал ошибок, и тип говорит правду.

Страницы потоком

Ночной отчёт читает весь архив. Старый код делал это циклом offset += 50, пока не придёт короткая страница, и цикл знал и про страницы, и про отчёт.

Stream.paginate разделяет. Он помнит состояние, тут это offset, и на каждом шаге просит страницу и следующее состояние. Option.none() вместо состояния значит «страниц больше нет»:

const PAGE = 50;

const allLetters = Stream.unwrap(
  Effect.map(Archive, (archive) =>
    Stream.paginate(0, (offset: number) =>
      Effect.map(
        archive.use((legacy) => legacy.list(offset, PAGE)),
        (page) => [page, page.length < PAGE ? Option.none() : Option.some(offset + PAGE)] as const,
      ),
    ),
  ),
);
// Stream<Letter, ArchiveError, Archive>

Stream.unwrap достаёт поток из эффекта: сначала берём сервис, потом строим на нём поток. Дальше это обычный поток из урока 85. Stream.take(allLetters, 10) прочтёт только первую страницу, вторую никто не попросит. Отказ на третьей странице это ArchiveError, а не половина отчёта, которую кто-то примет за целый.

События потоком

Новые письма архив отдаёт колбэком. В колбэк нельзя вернуть ошибку, его нельзя остановить и легко забыть от него отписаться. Stream.callback превращает подписку в поток:

const arrivals = Stream.unwrap(
  Effect.map(Archive, (archive) =>
    Stream.callback<Letter, ArchiveError>((queue) =>
      Effect.acquireRelease(
        archive.use(async (legacy) => legacy.subscribe((letter) => Queue.offerUnsafe(queue, letter))),
        (subscription) => Effect.sync(() => subscription.unsubscribe()),
      ),
    ),
  ),
);

Колбэк кладёт письмо в очередь потока. Подписка открыта через acquireRelease, поэтому отписка случится, когда поток перестанут читать: дочитали Stream.take(arrivals, 2), прервали файбер, упали. Все три случая закрывает один финализатор.

Граница: `ManagedRuntime`

Сервер Почтамта живёт на Hono, а Hono понимает промисы, не эффекты. Нужна граница, где эффект становится промисом.

Ты уже знаешь её по уроку 58: ManagedRuntime.make(layer) собирает слой один раз и исполняет на нём сколько угодно эффектов.

const runtime = ManagedRuntime.make(archiveLayer('magistral://archive'));

app.get('/letters/:id', (c) =>
  runtime
    .runPromise(findLetter(c.req.param('id')))
    .then((letter) => (letter === undefined ? c.text('нет письма', 404) : c.json(letter))),
);

Сто запросов, одно соединение с архивом. Когда сервер гасят, runtime.dispose() разбирает слой, и финализатор закрывает архив.

Работает, только архив открывается и закрывается на каждый запрос. Сто горожан у окошка значит сто соединений, и старый клиент, который рассчитан на одно, начнёт отказывать к обеду. Слой это рецепт, а Effect.provide готовит по нему заново каждый раз. Рантайм готовит один раз.

Беда при сборке

Сделаем обработчик честнее: беда архива должна быть ответом 503, а не упавшим запросом. Первая мысль: Effect.catchTag('ArchiveError', ...) внутри runtime.runPromise. Пока архив открылся, это работает. Но если архив заперт уже при сборке слоя, catchTag не сработает: рантайм падает раньше, чем запускает твой эффект. runPromise отказывает с ArchiveError, и Hono отвечает 500.

Лечится тем, что сборку слоя делают частью эффекта. У рантайма есть contextEffect: эффект, который отдаёт собранный контекст или падает ошибкой сборки. Собираем запрос целиком:

const respond = (effect: Effect.Effect<Response, ArchiveError, Archive>) =>
  Effect.runPromise(
    runtime.contextEffect.pipe(
      Effect.flatMap((context) => Effect.provide(effect, context)),
      Effect.catchTag('ArchiveError', (error) =>
        Effect.succeed(Response.json({ error: error.message }, { status: 503 })),
      ),
    ),
  );

Теперь catchTag видит обе беды: и запертый при сборке архив, и сгоревшую страницу при запросе. Слой при этом всё так же строится один раз: contextEffect отдаёт уже собранный контекст.

Под капотом

Рантайм держит три вещи: свою область (scope), таблицу собранных слоёв и результат сборки. Сборка идёт один раз, при первом запуске, и её результат кэшируется вместе с ошибкой. Архив открыли через минуту, а contextEffect всё равно отдаёт ту же ArchiveError: новой попытки не будет. Чинится это снаружи: смотритель, который видит 503, создаёт новый рантайм, а старый разбирает через dispose.

dispose закрывает область рантайма. Финализаторы слоя исполняются в обратном порядке, и архив получает свой end(). Если сервер гасят по сигналу, dispose зовут в обработчике сигнала, иначе финализаторы не успеют.

Hono по одному окошку

Теперь сам переход. В старом сервере десяток маршрутов, все на промисах:

const app = new Hono();

app.get('/letters/:id', async (c) => {
  const archive = await openArchive(URL);
  try {
    const letter = await archive.find(c.req.param('id'));
    return letter === undefined ? c.text('нет письма', 404) : c.json(letter);
  } finally {
    await archive.end();
  }
});
// и ещё девять маршрутов

Шаг первый: один маршрут. /letters/:id зовёт respond, остальные девять не меняются. Это уже можно выкатывать: если что-то пошло не так, откатывается один маршрут, а не весь сервер.

app.get('/letters/:id', (c) =>
  respond(
    Effect.map(findLetter(c.req.param('id')), (letter) =>
      letter === undefined
        ? new Response('нет письма', { status: 404 })
        : Response.json(letter),
    ),
  ),
);

Шаг второй и дальше: маршрут за маршрутом. Новый код зовёт сервисы, и прямых вызовов openArchive становится меньше.

Шаг последний: когда под префиксом не осталось промисов, весь префикс отдают одному обработчику fetch на Effect. В Hono для этого есть app.mount:

const archiveApp = makeHandler(archiveLayer(URL)); // обработчик из разминки: { fetch, dispose }
app.mount('/archive', archiveApp.fetch);

mount отрезает префикс: запрос на /archive/letters/L-3 придёт в обработчик как /letters/L-3. Обработчик принимает обычный Request и отдаёт Response, поэтому один и тот же fetch повесится и в Hono, и в Bun, и в Deno. Когда своих маршрутов у Hono не останется, его уберут, и горожане не заметят.

Важно. Hono в браузерном раннере есть, но шаг Т не знает его типов, поэтому разминка собирает обработчик fetch без Hono. В Hono он вешается строкой app.all('*', (c) => handler.fetch(c.req.raw)).

Разминка

Оберни старую книгу башен Магистрали: соединение в слое, дверь use, смены и страницы потоком, обработчик fetch на ManagedRuntime с 404 на чужой путь и 503 на любую беду книги, даже при сборке.

EF-118-1★

Архив отдаёт ровно 100 писем страницами по 50. Сколько раз allLetters позовёт list, если дочитать поток до конца? А если условие конца сменить на «пустая страница»?

EF-118-2★★

Рантайм собрался с ошибкой «архив заперт», а через минуту архив открыли. Обработчик всё ещё отвечает 503. Напиши, как обработчику самому пересоздать рантайм после ошибки сборки, не трогая рантайм, который собрался удачно. Где окажется dispose старого рантайма?

Практика

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

Как переводить на новое целые сервисы, а не только маршруты, и чем Strangler Fig отличается от Anti-Corruption Layer, разобрано в уроке про микросервисы раздела про бэкенд.

Дальше

К вечеру справочная Почтамта ходит в архив через одну дверь. Девять маршрутов из десяти ещё на промисах, но соединение с архивом одно, а запертый архив горожанин видит как честное «архив недоступен», а не как пустую страницу.