Проверка буфера и завершения AsyncStream на облачном Mac

Если передавать в AsyncStream данные о ходе загрузки, состоянии устройства или уведомления из колбэков, корректное обновление интерфейса ещё не означает, что поток событий работает правильно. Стоит потребителю немного отстать — и буфер может начать терять значения; после закрытия страницы подписка на источник событий может продолжить работать. На облачном Mac VpsKeep можно сначала закрепить эти правила поведения отдельными тестами на Swift, а затем подключить поток к приложению, не выискивая гонки многократными нажатиями в интерфейсе.

Определите, какие события можно отбрасывать

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

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

Вызов yield у AsyncStream не ждёт, пока потребитель освободит место. Ограниченный буфер задаёт стратегию отбрасывания, а не обеспечивает автоматическое обратное давление.

В этой статье правила проверяются на потоке целых чисел с ёмкостью буфера 2. Числа обозначают порядковые номера событий, а не результаты измерения производительности.

Подготовьте отдельный проект для тестов

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

mkdir StreamContract
cd StreamContract
swift --version
swift package init --type library
mkdir -p Tests/StreamContractTests

Сохраните код из следующего раздела в Tests/StreamContractTests/StreamContractTests.swift. Если при инициализации уже создан файл с таким именем, замените его. Остальные тесты, созданные шаблоном, можно оставить: ниже запускаются только тесты этого класса.

Чтобы результат переполнения был детерминированным, тест сначала синхронно создаёт три элемента и только затем запускает потребление. Не позволяйте производителю и потребителю конкурировать за выполнение и не пытайтесь угадать, заполнен ли буфер, с помощью «ожидания в одну секунду». Здесь проверяется контракт контейнера, а не скорость выполнения потоков.

Проверьте и отброшенное значение, и сохранённую последовательность

Проверки итогового массива недостаточно. Код также должен проверять результат третьего вызова yield: при сохранении новейших элементов отбрасывается старый элемент 1, а при сохранении самых ранних — новый элемент 3. После finish значения, уже попавшие в буфер, по-прежнему можно прочитать до конца.

import XCTest

final class StreamContractTests: XCTestCase {
    func testNewestKeepsLastTwo() async {
        let pair = AsyncStream<Int>.makeStream(
            bufferingPolicy: .bufferingNewest(2)
        )
        pair.continuation.yield(1)
        pair.continuation.yield(2)

        switch pair.continuation.yield(3) {
        case .dropped(let value):
            XCTAssertEqual(value, 1)
        default:
            XCTFail("Expected a dropped value")
        }

        pair.continuation.finish()
        var received: [Int] = []
        for await value in pair.stream {
            received.append(value)
        }
        XCTAssertEqual(received, [2, 3])

        switch pair.continuation.yield(4) {
        case .terminated:
            break
        default:
            XCTFail("Expected a terminated stream")
        }
    }

    func testOldestKeepsFirstTwo() async {
        let pair = AsyncStream<Int>.makeStream(
            bufferingPolicy: .bufferingOldest(2)
        )
        pair.continuation.yield(1)
        pair.continuation.yield(2)

        switch pair.continuation.yield(3) {
        case .dropped(let value):
            XCTAssertEqual(value, 3)
        default:
            XCTFail("Expected a dropped value")
        }

        pair.continuation.finish()
        var received: [Int] = []
        for await value in pair.stream {
            received.append(value)
        }
        XCTAssertEqual(received, [1, 2])
    }
}

В рабочем коде при обработке .dropped следует увеличивать счётчик отброшенных событий или запускать предусмотренный бизнес-логикой механизм компенсации, а не синхронно записывать в журнал каждое событие прогресса. При получении .terminated нужно прекратить передачу значений, а не бесконечно повторять попытки с тем же continuation.

Убедитесь, что отмена вызывает обработчик отписки

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

extension StreamContractTests {
    func testCancellationReportsTermination() async {
        let ended = expectation(description: "Stream cancelled")
        let pair = AsyncStream<Int>.makeStream()

        pair.continuation.onTermination = { reason in
            switch reason {
            case .cancelled:
                ended.fulfill()
            case .finished:
                XCTFail("Unexpected normal finish")
            @unknown default:
                XCTFail("Unknown termination reason")
            }
        }

        let consumer = Task {
            for await _ in pair.stream {}
        }

        consumer.cancel()
        await consumer.value
        await fulfillment(of: [ended], timeout: 2)
    }
}

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

Нужно учесть и гонку при создании подписки: завершение может произойти раньше, чем будет возвращён токен. Адаптер должен хранить состояние «закрыт». Если токен приходит после закрытия, его необходимо сразу освободить. Одной установки onTermination недостаточно: без обработки этого сценария освобождение подписки можно пропустить.

Подключите тесты к пайплайну и проверьте готовность к интеграции

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

swift test --filter StreamContractTests

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

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

Часто задаваемые вопросы