Краткий ответ

Нет. Политика bufferingNewest(1) не создаёт обратного давления: производитель не приостанавливается в ожидании потребителя. Если буфер заполнен, новый элемент принимается, а самый старый ещё не прочитанный элемент отбрасывается.

Исторический контекст

AsyncStream предназначен для адаптации событийных и callback-интерфейсов к модели async/await и AsyncSequence. В таких источниках производитель часто работает независимо от скорости потребителя, поэтому библиотеке нужен явный способ определить поведение при переполнении буфера.

Политика буферизации разделяет две задачи: доставку элементов и управление памятью. Она не превращает поток в синхронный канал и сама по себе не заставляет производителя подстраиваться под потребителя.

Постановка проблемы

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

При bufferingNewest(1) в буфере сохраняется только последнее доступное значение. Ошибка в ожиданиях возникает, если разработчик считает, что каждый вызов yield обязательно будет доставлен: при переполнении промежуточные элементы теряются.

Подробное решение

bufferingNewest(1) ограничивает количество ожидающих элементов одним. Когда потребитель успевает забрать значение, новый элемент помещается в свободный буфер. Когда буфер заполнен, новый элемент сохраняется, а старый удаляется.

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

func makeStream() -> AsyncStream<Int> {
    AsyncStream(bufferingPolicy: .bufferingNewest(1)) { continuation in
        Task {
            for value in 1...3 {
                let result = continuation.yield(value)
                print(value, result)
            }
            continuation.finish()
        }
    }
}

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

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

Политика буферизации также не делает сами элементы безопасными для конкурентной передачи. Если элементы содержат изменяемые ссылочные объекты, безопасность их доступа должна обеспечиваться отдельно, например через Sendable, actor или другой механизм синхронизации.

Ситуация из практики

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

Для экрана выбран bufferingNewest(1): ему нужно актуальное положение, а не полный журнал перемещений. Потеря промежуточных координат допустима, производитель не блокируется, память ограничена, а задержка между фактическим и отображаемым состоянием остаётся небольшой.

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

Что кандидаты часто упускают

Означает ли bufferingNewest(1), что потребитель всегда получит самое последнее значение?

Нет, это не абсолютная гарантия на любой момент времени. Политика сохраняет самое новое значение среди тех, которые находятся в буфере, но потребитель может получить элемент до появления более свежего. Кроме того, после завершения потока новых значений уже не будет.

Можно ли использовать AsyncStream с ограниченным буфером для доставки обязательных событий?

Только если допустима потеря событий согласно выбранной политике. bufferingNewest и bufferingOldest являются стратегиями отбрасывания, а не механизмом надёжной доставки. Для обязательных событий необходимо выбрать архитектуру, в которой переполнение приводит к ожиданию, ошибке, сохранению во внешнем хранилище или другому явно обработанному результату.

Что произойдёт с производителем после завершения AsyncStream?

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