Правила накопления

Страница «Накопление событий до запуска задачи» даёт четыре рецепта. Эта страница даёт правила, на которых они стоят: когда группа готова, к какой группе присоединяется событие и как задача группы начинается, отменяется, падает и заканчивается на close(). Код берёт два контроллера рецептов: SettingsController из раздела «Настройки, сохраняемые на каждом щелчке» и LogController из раздела «Один запрос на строку журнала».

Строки под кодом — то, что код печатает.

Выбор момента готовности группы

Обе фабрики принимают необязательный timing. Настройка определяет, когда группе можно стартовать; collect сохраняет каждое принятое событие, а accumulate сохраняет результат своей merge. Например, merge: (accumulated, incoming) => incoming сохраняет последнее значение.

	debounce(duration)	throttle(duration)	throttle(duration, startAtOnce: false)
Первая группа готова	через duration без новых событий	сразу	через duration
Добавление	перезапускает таймер	не продлевает таймер	не продлевает таймер
Ожидание отсчитывается от	последнего принятого события	прошлого фактического старта	прошлого старта или появления группы

AccumulationTiming.debounce(duration) ждёт паузы после последнего принятого события в каждой группе. Каждое добавление перезапускает таймер, даже если merge вернула неизменное значение. При окне 200 мс события на 0, 60 и 120 мс делают группу готовой на 320 мс. Когда таймер debounce срабатывает, группа запечатывается, даже если другая Job работает. Следующие события создают новую группу и не меняют запечатанный вход. collect делает снимок в этот момент и больше его не повторяет. Непрерывный вход может удерживать открытую группу неограниченно долго. Ради этого ожидания debounce и стоит в рецепте поиска: запрос уходит, когда набор замер. Из-за него же сборщик журнала берёт throttle, а не debounce: записям, которые должны дойти все, нельзя ждать паузы, которой может и не быть.

AccumulationTiming.throttle(duration) разрешает первый старт сразу, затем выдерживает не менее duration между фактическими стартами групп одного накопителя. Добавления за этот интервал собираются в открытую группу, не продлевая таймер. По истечении интервала накопленное готово без нового события. Группа принимает события, пока очередь не возьмёт её на выполнение. Интервал, за который событий не было, Job не создаёт, и группа следующего события готова сразу: интервал отсчитывается от прошлого старта и уже истёк.

startAtOnce: false отсчитывает интервал и до первой группы. Когда у накопителя появляется группа, чего она ждёт, зависит от того, что у накопителя уже есть:

• Ничего своего в очереди и в работе и ни одного идущего интервала: группа заводит интервал в момент появления и готова по его истечении, вместе со всем, что записали тем временем.
• Ничего своего в очереди и в работе, но ещё идёт интервал от прошлого старта: группа готова, когда он кончится. Уже идущий интервал не перезаводится никогда.
• Другая своя группа, которая ждёт в очереди или работает: новая группа интервала не заводит и ждёт тот, который возобновит старт другой группы.

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

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

У ожидания есть своя цена. Единственное событие на простаивающем накопителе задержано на целый интервал, и close(mode: SoloCloseMode.drain) ждёт вместе с ним, а обычный close() его выбрасывает. Берите startAtOnce: false там, где частота важнее задержки первого события, как в рецепте журнала, и оставляйте умолчание там, где именно первого события и ждут.

• Что расходует интервал. Стартом считается переход к выполнению, до onStart. Группа, отклонённая стартовыми правилами, не расходует интервал throttle; отмена из onStart расходует. При startAtOnce: false после отказа у накопителя не остаётся ничего в очереди, поэтому следующая появившаяся группа отсчитывает свой полный интервал от собственного появления.
• Слот, занятый другой задачей. Если группа стартовала на 0 мс с интервалом 200 мс, а другая Job занимает слот до 500 мс, следующая группа стартует на 500 мс, а последующая не раньше 700 мс. Если собственный обработчик группы, дети или уборка длятся дольше интервала, следующая группа может стартовать сразу после их завершения.
• Что показывает очередь. Ожидающие группы видны в queue.jobs. Очередь берёт первую готовую Job в порядке списка, пропуская ожидающие группы. Поэтому список может быть непустым, когда ни одна Job не работает. Ожидание времени не занимает слот выполнения; после старта обработчика очередь ждёт его детей и уборку перед запуском следующей корневой Job.
• Нулевая и отрицательная длительность. Без timing или при Duration.zero группы готовы сразу и таймеров нет, что бы ни говорил startAtOnce. Отрицательная длительность бросает ArgumentError.
• Таймеры. Общую настройку времени можно передать нескольким накопителям; у каждого свои группы и интервал throttle. Колбэки таймеров следуют порядку событий Dart: событие, обработанное до колбэка debounce, может перезапустить таймер; событие после него относится к новой группе. Занятый цикл событий может задержать колбэки и старты.

Выбор места присоединения событий

// Три вызова подряд: очередь между ними ничего не берёт. Куда ляжет A2,
// решает политика.
final a1 = settings.update(const SettingsPatch(theme: 'dark'));
settings.reload(); // B, обычная отдельная Job
final a2 = settings.update(const SettingsPatch(language: 'ru'));

Обе фабрики принимают AccumulationPolicy, фиксируемую при создании накопителя.

Политика	Очередь после добавлений	Хэндл для A2
adjacent	[A1, B, A2]	Новая задача
replace	[B, A(A1 + A2)]	Существующая задача A1, перенесённая за B
join (по умолчанию)	[A(A1 + A2), B]	Существующая задача A1

adjacent пристраивает событие к группе, только если та стоит в хвосте очереди. replace переставляет задачу группы в хвост: находит последнюю открытую ожидающую группу того же накопителя, проходя через другие задачи, и переносит её задачу за них, так что A оказывается за B. join присоединяет событие к той же группе там, где она стоит, и A сохраняет место перед B. В обоих случаях B остаётся в очереди: поиск её только пропускает. Выполнение также зависит от готовности: готовая B может пройти перед A, пока A ждёт своего времени.

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

Берите adjacent там, где эта граница и есть суть:

• где merge выбрасывает то, что заменяет, и задача, вставшая между двумя событиями, обязана выполниться между ними, как в рецепте команд;
• где такая задача меняет смысл накопленного.

Политики смотрят на текущую очередь. Если B уже выполнилась, ожидающая A снова может оказаться хвостом, и следующее событие присоединится к ней при adjacent. Уже раздельные группы задним числом не объединяются. Запечатанная debounce-группа не подходит ни одной из трёх политик.

replace переставляет задачу найденной группы в хвост, добавив в эту группу входящее событие. Хэндл остаётся тем самым, который получили все добавления в эту группу, принятые события остаются в ней, а группа, уже стоящая хвостом, сохраняет своё место. По дороге ничего не отменяется, поэтому ни один колбэк отмены не выполняется посреди переноса. Solo.debug, собственная трасса движка из раздела «Логи» страницы о диагностике, печатает move <задача> to the tail; события наблюдателя нет, потому что ни одна задача не стартовала и не завершилась.

Перенос начинает новое окно debounce и сохраняет интервал throttle накопителя. Он применим к ожидающей группе, в том числе настроенной с cancellable: false: перенос не удаление, и флагу нечего отклонять. Работающая группа не переносится: она перестаёт принимать события, когда очередь её забрала. У обычной политики очереди Policy.keepLatest свои правила отмены; AccumulationPolicy объявлен отдельным enum.

Возможность объединения определяется идентичностью накопителя. Два накопителя с одинаковым key остаются разными. Ключ по-прежнему помечает их задачи для обычного поиска и политик очереди. Создание накопителя заново на каждом вызове не даёт событиям присоединяться к существующему.

Запуск, отмена и ошибки

final group = logs.logEvent(const LogEntry('checkout opened'));
final same = logs.logEvent(const LogEntry('cart shown'));

// Все, кто добавлял в эту группу, держат один и тот же хэндл...
print(identical(group, same));

// ...поэтому его отмена отменяет всю группу, и каждый из них читает
// один и тот же исход.
await group.cancel();
switch (await same.done) {
  case Done():
    print('sent');
  case Failed(:final error):
    print('failed: $error');
  case Cancelled(:final reason):
    print('cancelled: $reason');
}

true
cancelled: manual

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

По той же причине простой then на задаче группы бросает StateError, и на первом событии тоже. Каждый вызывающий продолжает задачу через shared(), group.shared().then(...), и отмена того, что вернул shared(), кончает только эту ветку. См. раздел «Ручка на каждую ветку».

Все три политики работают с ожидающими группами. Группа прекращает приём событий, когда её запечатывает debounce или когда очередь её берёт, до canStart и onStart. События из этих колбэков идут в последующую группу. Обработчик выполняется с заданным рабочим типом и правилами, а задача группы во всём остальном для очереди обычная Job.

До первого события накопитель не создаёт задач. Без timing, если каждая группа завершается до прихода следующего события, каждое событие запускает отдельную задачу. Поиск при join может пройти всю очередь, как и при replace, а adjacent проверяет только её хвост.

merge должна быть синхронной и чистой:

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

canStart, keepWhile и отмена действуют на всю задачу, а сами правила описаны в разделе «Состояние и правила» страницы о состоянии. Стартовое правило, отклонившее задачу, отменяет всё, что группа накопила. Результат и ошибки обработчика следуют обычному контракту Job; принятая отмена сохраняет приоритет над более поздним значением или ошибкой. Накопитель не откатывает частичные внешние эффекты и не пересылает неудачный вход сам.

Рецепт «Настройки, сохраняемые на каждом щелчке» предполагает одного писателя и future API, которая завершается при фактическом окончании записи. ctx.join удерживает слот выполнения до её завершения, в том числе после отмены, и этим держится то, что следующая запись не начнётся поверх этой. Один клиентский таймаут не доказывает, что сервер перестал записывать. save, завершающая свою future по таймауту, возвращает слот, пока сервер ещё пишет, и следующая группа отправляет свою запись поверх незакончившейся. Какая из двух уцелеет, решает тогда уже сервер, а не очередь. Экран и сервер могут разойтись насовсем. Отмена также может помешать финальному emit после того, как сервер уже принял запись.

Срок, заданный аккумулятору через timeout, работает как отмена, а не как клиентский таймаут: ctx.join держит слот, пока save не завершится, поэтому следующая запись по-прежнему ждёт конца этой. Срок обходится в финальный emit. Группа, у которой вышел срок, отменяется, пока сервер ещё пишет, кончается Cancelled(timeout), когда запись сохранена, и до своего emit не доходит. Состояние сохраняет старые настройки, и следующая группа накладывает свой патч на них и записывает старые значения поверх изменения.

Группа, которая пишет собственное состояние, пока работает, оставляет его после себя, когда падает или её отменяют: после принятой отмены emit бросает, и close() посреди выгрузки оставляет Uploading(3) навсегда. Обработчики состояния ifFailed и ifCancelled откатывают его так же, как у job. Это не обработчик группы: группа вызывает один из них не больше одного раза, сколько бы событий в ней ни было, после своего обработчика, своих детей и своей уборки. Таблица «Допуск обработчиков и их ошибки» страницы о состоянии действует на группу так же, как на любую задачу:

sealed class UploadState {
  const UploadState();
}

final class Waiting extends UploadState {
  const Waiting();
}

final class Uploading extends UploadState {
  final int photos;

  const Uploading(this.photos);
}

final class UploadFailed extends UploadState {
  final Object error;

  const UploadFailed(this.error);
}

final class PhotoUploader extends Solo<UploadState> {
  final Future<void> Function(List<String> photos) _send;
  late final _photos = collect<UploadState, String, void>(
    (ctx, photos) async {
      ctx.emit(Uploading(photos.length));
      await ctx.abandonable(() => _send(photos));
      ctx.emit(const Waiting());
    },
    key: 'upload',
    timing: AccumulationTiming.debounce(const Duration(milliseconds: 200)),
    ifFailed: (state, error, stackTrace) => UploadFailed(error),
    ifCancelled: (state, cancelled) => const Waiting(),
  );

  PhotoUploader(this._send) : super(const Waiting());

  Job<void> addPhoto(String path) => _photos.add(path);
}

uploading: Uploading(3)
after close(): Waiting
a failed upload: UploadFailed(Bad state: offline)

Группа, которая так и не началась, не вызывает ни одного обработчика состояния: отменённая в своём окне или пока ждёт в очереди, убранная clear, ещё ждущая, когда контроллер закрывается, или отклонённая canStart. Группа, которую canStart пропустил, уже начата, даже до вызова её обработчика: отмена в onStart вызывает ifCancelled, а обработчик так и не видит событий.

Отмена ожидающей debounce-группы убирает её таймер. Удаление или очистка throttle-групп сохраняет уже начатый интервал, поэтому новое событие не обходит ограничение. Таймер истекает один раз; он заводится заново, когда группа фактически стартует, а при startAtOnce: false ещё и когда группа появляется на накопителе, у которого ничего своего не стоит в очереди и не работает.

close() отменяет все таймеры timing, снимает ожидающие группы и отменяет или ждёт работающую задачу по обычным правилам. Последнюю пачку он не отправляет: группа, которая ещё ждёт своего окна, пропадает, а её задача, единственный хэндл всех добавлявших, завершается Cancelled(closed). Второй выбор, close(mode: SoloCloseMode.drain), дожидается окна накопления и выполняет уже стоящие в очереди группы по правилам из раздела «Отмена работы и закрытие контроллера» страницы об отмене. Ничего нового не принимает ни тот, ни другой: вызов add после close() возвращает новую задачу, уже завершённую Cancelled(closed); merge и обработчик не вызываются. Так и при дренаже, для группы, которая ещё ждёт своего окна: событие — новые данные, поэтому в группу оно не входит, и группа уходит такой, какой была на close. Вызов под Policy.droppable при дренаже получает свою задачу из очереди потому, что работы не добавляет, а событие добавило бы. Всё, что должно попасть в последнюю пачку, добавляют до close. После завершения группы её внутреннее хранилище входа освобождается. Обработчик, результат или ошибка, сохранившие вход, по-прежнему владеют своими ссылками.