Раздел 46 · Effect: Тихая Академия

Мёртвые письма

middle~25 мин

без входаВойди, чтобы отмечать прогресс, вести заметки и решать задачи в редакторе. войти

frames из урока 85 выбрасывает мусор молча: строка, которая не разобралась, исчезает, и утром никто не скажет, сколько их было и что в них было. Липвиг прав, и у его комнаты есть имя: очередь мёртвых писем. Сегодня башня перестаёт выбрасывать строки, а заодно узнаёт, куда вообще течёт поток и как устроены его трубы.

Почему не ловить ошибку потока

Первая мысль: пусть разбор будет обычным эффектом в mapEffect, отказ станет ошибкой потока, а мы её поймаем в конце. Так нельзя, и важно понять почему. Ошибка потока заканчивает поток. Как у эффекта: после fail дальше ничего не исполняется. Поймать её можно, но поток уже кончился, и все строки после битой не прочитаны. Одна строка мусора в полночь, и вся ночь до утра не обработана.

Значит, отказ разбора не ошибка потока, а значение: ещё одно, что может прийти по линии. Как в уроке 42, где ошибку превращали в Result и шли дальше:

export interface DeadLetter {
  readonly line: number;
  readonly text: string;
  readonly reasons: readonly string[];
}

export const sortLines = <E, R>(
  lines: Stream.Stream<string, E, R>,
): Stream.Stream<Result.Result<FrameV2, DeadLetter>, E, R> =>
  lines.pipe(
    Stream.zipWithIndex,
    Stream.mapEffect(([text, index]) =>
      readLine(text).pipe(
        Effect.mapError((reasons): DeadLetter => ({ line: index + 1, text, reasons })),
        Effect.result,
      ),
    ),
  );

Stream.zipWithIndex даёт каждой строке её номер с нуля, бирке нужен номер с единицы. readLine это разбор одной строки так же, как в nightIntake из урока 84: версия по полю version, беды словами. Effect.result превращает отказ в значение Result.fail, и поток идёт дальше. Ошибка E самой линии, обрыв провода, остаётся ошибкой: это не письмо, а авария, и её прятать нельзя.

Теперь развести кадры и письма. Кадры идут дальше по потоку, письма кладут в очередь из урока 68, которую разбирает кто-то другой:

export const withDeadLetters = <E, R>(
  lines: Stream.Stream<string, E, R>,
  dead: Queue.Enqueue<DeadLetter>,
): Stream.Stream<FrameV2, E, R> =>
  sortLines(lines).pipe(
    Stream.mapEffect((result) =>
      Result.isSuccess(result)
        ? Effect.succeed(Option.some(result.success))
        : Effect.as(Queue.offer(dead, result.failure), Option.none()),
    ),
    Stream.filter(Option.isSome),
    Stream.map((frame) => frame.value),
  );

Queue.Enqueue это очередь, в которую можно только класть: функции не нужно право из неё брать, и тип это говорит. Queue.offer здесь эффект, а не offerUnsafe из урока 86. Если очередь мёртвых писем ограничена и полна, поток подождёт, пока её разберут, а не потеряет письмо: здесь ждать есть кому. И письмо лежит в очереди раньше, чем следующий кадр ушёл дальше: mapEffect по одному сохраняет порядок событий, а не только значений.

Куда течёт поток: Sink

До сих пор поток кончался в runCollect или runForEach. Это два частных случая одной вещи: сток, Sink. Поток описывает, откуда берутся значения, сток описывает, куда они уходят и что из этого выйдет. Stream.run(stream, sink) соединяет одно с другим и даёт эффект.

Sink<A, In, L, E, R> читается так: принимает значения In, в конце даёт A, может вернуть непрочитанный остаток L (об этом ниже), может упасть с E, просит сервисы R. runCollect это run(Sink.collect()), runForEach(f) это run(Sink.forEach(f)).

Зачем отдельное значение, если есть run*? Затем же, зачем схема, а не функция разбора: сток можно назвать, положить в модуль, переиспользовать и проверить отдельно от потока. Почтамту нужна сводка: сколько кадров каждого вида прошло за ночь.

export type Tally = { readonly [K in Message['kind']]: number };

export const tally: Sink.Sink<Tally, FrameV2> = Sink.reduce(
  (): Tally => ({ text: 0, ack: 0, ping: 0, lost: 0 }),
  (counts, frame) => ({ ...counts, [frame.message.kind]: counts[frame.message.kind] + 1 }),
);

Sink.reduce это Array.prototype.reduce для потока: начальное значение лениво, функцией, чтобы каждый запуск начинал с нуля. Тот же tally годится для ночи, для одной линии и для теста, где поток из трёх кадров: Stream.run(frames, tally).

Сток видит поток кусками, как throttle в прошлом уроке. Большинству стоков это безразлично, а архиву это подарок: писать на диск куском дешевле, чем по строке.

export const archive = <E, R>(
  write: (frames: readonly FrameV2[]) => Effect.Effect<void, E, R>,
): Sink.Sink<void, FrameV2, never, E, R> => Sink.forEachArray(write);

Sink.forEachArray зовёт write с каждым куском потока. Пять кадров кусками по два это три записи: два, два и один. Если запись упала, поток останавливается: сток не просит следующий кусок, и линия дальше не читается. Противодавление работает и тут.

Остаток

L в типе стока нужен для стоков, которые кончаются раньше потока. Sink.take(3) берёт три значения и заканчивает, а если кусок был из пяти, два непрочитанных значения он возвращает как остаток. Остаток не теряется: Stream.transduce(stream, Sink.take(3)) прогоняет поток через один и тот же сток раз за разом, и каждый новый проход начинается с остатка прошлого. Так из стока получаются пачки. В задаче остаток не нужен, но в типах он будет попадаться, и теперь понятно, что это.

Под капотом

Табло на площади: горячий поток

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

Все потоки до сих пор были холодными: каждый запуск это новая подписка и всё с начала. Табло это горячий поток: оно живёт само, а зрители приходят и уходят. Для этого есть SubscriptionRef, ячейка Ref из урока 65, на изменения которой можно подписаться:

const passed = yield* SubscriptionRef.make(0);

export const counting =
  (passed: SubscriptionRef.SubscriptionRef<number>) =>
  <A, E, R>(self: Stream.Stream<A, E, R>): Stream.Stream<A, E, R> =>
    Stream.tap(self, () => SubscriptionRef.update(passed, (count) => count + 1));

const board = SubscriptionRef.changes(passed); // Stream<number>

Stream.tap делает эффект с каждым значением и пропускает значение дальше нетронутым: кадры идут своей дорогой, табло прибавляет единицу. SubscriptionRef.changes(passed) это поток, который сначала отдаёт текущее значение, а потом каждое новое. Зритель, который подписался до полуночи, увидит 0, 1, 2, 3. Зритель, который подошёл после третьего кадра, увидит сразу 3 и дальше по порядку.

Где тут дыра

const night = lines.pipe(
  Stream.mapEffect((line) => Schema.decodeUnknownEffect(Incoming)(line)),
  Stream.runForEach(relay),
  Effect.catch((error) => Effect.log(`мусор: ${error.message}`)),
);

На его проверке мусор был последней строкой ночи: все кадры ушли, в логе одна запись, всё сходится. Этой ночью мусор пришёл второй строкой. В логе одна запись, мусор: ..., и больше ничего: ни одного кадра после второй строки башня не отправила. Ошибка в mapEffect закончила поток, а Effect.catch поймал её уже после конца. Ночь не упала, она кончилась в полночь. И в логе не бирка, а строчка, которую никто не разберёт: ни номера строки, ни самой строки.

Магистраль, этап 13

Башня заводит комнату мёртвых писем: каждая строка становится кадром или письмом с биркой, письма уходят в очередь. Ещё два стока, сводка и архив, и табло, которое горожане смотрят на площади. В version.ts снизу ложится readLine: разбор одной строки, как в nightIntake.

Практика

Дальше

Комната мёртвых писем башни к утру набирает девять бирок, и у каждой есть номер строки. А в сводке tally Пондер находит странное: подтверждений ack втрое больше, чем отправленных кадров. Ринцвальд, который весь вечер молча сидел в углу, наконец говорит, тихо и не глядя ни на кого.

Заказ на ретранслятор ждёт в следующем уроке.