クラウド Mac で検証する AsyncStream のバッファと終了処理

ダウンロードの進捗、デバイスの状態、通知のコールバックを AsyncStream に接続しても、画面が正常に更新されているだけでは、イベントストリームが正しく動作しているとは限りません。コンシューマーの処理が少し遅れるだけで、バッファから値が失われることがあります。画面を離れた後も、イベントソースの購読が動き続ける可能性があります。VpsKeep のクラウド Mac では、まず独立した Swift テストでこうした動作を明確に検証してから、アプリに組み込めます。画面を何度も操作して競合状態を探す必要はありません。

破棄してよいイベントを先に決める

進捗率は通常、最新の値だけあればよいため、bufferingNewest が適しています。先に届いたイベントを残す必要がある場合は、bufferingOldest を選べます。ただし、どちらも値の欠落を防ぐキューではありません。注文の変更やファイルの書き込み完了通知など、失ってはいけないデータには、別途、確認応答や永続化の仕組みが必要です。

デフォルトの無制限バッファも万能ではありません。生成速度が消費速度を長時間上回ると、キューにたまった要素がメモリを占有し続けます。まず、バッファの容量、満杯になったときにどの要素を残すか、いつ購読を解除するか、という三つの条件を書き出してください。容量は、業務上許容できる滞留量に基づいて決めます。マシンの空きメモリに合わせて、むやみに大きくするものではありません。

AsyncStream の yield は、コンシューマーが空きを作るまで待機しません。容量を制限したバッファが提供するのは破棄ポリシーであり、自動的なバックプレッシャーではありません。

この記事では、バッファ容量が 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 に対して再試行を繰り返してはいけません。

キャンセルで購読解除の処理が呼ばれることを検証する

同じファイルに、次の拡張を追加します。ストリームには要素がないため、コンシューマーは待機するか、キャンセルに応答するしかありません。終了通知はテストの expectation で検証し、固定時間のスリープには依存しません。二秒はテスト失敗を判定するためのタイムアウトであり、実行手順の間に挟む待ち時間ではありません。

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)
    }
}

実際のアダプターでは、expectation を満たす処理を、購読トークンを解放するメソッドに置き換えます。終了処理がメインスレッドで実行される保証はないため、そこから直接 UI を更新してはいけません。購読解除はスレッドセーフかつ冪等である必要があります。また、アダプター内部のロックを保持したまま、アダプター側のコールバックを再び呼び出す可能性がある購読解除関数を実行しないでください。

購読を確立する際の競合も考慮が必要です。終了が先に発生し、その後でトークンが返される場合があります。アダプターは「クローズ済み」の状態を保持しなければなりません。遅れて届いたトークンは、その状態を確認したら直ちに解放します。onTermination を設定するだけで、この経路の処理を見落とさないようにしてください。

パイプラインへの組み込みと引き渡し前の確認

テストを実行し、出力をパイプラインの通常のテスト記録に保存します。入力を固定しているため、繰り返し実行しても、得られる配列と破棄される要素は同じになるはずです。たまにしか成功しない場合は、実際のタイマーや共有のイベントソースがテストに混ざっていることが多くあります。

swift test --filter StreamContractTests

アプリに組み込む前に、四つの確認を済ませてください。通常の finish で購読が解除されるか、タスクのキャンセルで購読が解除されるか、購読解除中に遅れて届いたコールバックが無視されるか、終了後にトークンが届いた場合に直ちに解放されるか、という点です。まず手動でイベントを送れる偽のイベントソースでこれらの経路をテストし、その後で実際のソースをテストします。

同じストリームを複数のタスクで競合して読み取ることで、ブロードキャストを再現しようとしてはいけません。複数の購読者が必要なら、明示的に配信層を設計し、購読者ごとにバッファリングポリシーを定めます。最後に、テストコード、ツールチェーンのバージョン、業務上の契約を記した説明を残してください。後からイベントソースを接続する人が、どの破棄は設計どおりで、どの破棄は障害なのかを判断できるようにします。

よくある質問