雲端 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 重試。

驗證取消會觸發解除訂閱的入口

在同一個檔案中加入下列擴充。串流沒有任何元素，消費端只能等待或回應取消。終止通知透過測試期望來驗證，不依賴固定的休眠時間。兩秒只是判定失敗的逾時上限，不是執行步驟之間的等待時間。

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 是否解除訂閱；任務取消是否解除訂閱；解除訂閱期間較晚抵達的回呼是否被忽略；權杖晚於終止抵達時是否立即釋放。先用可手動投遞事件的假事件來源測試這些路徑，再測試真實來源。

不要讓多個任務競爭讀取同一個串流來模擬廣播。需要多個訂閱者時，應明確設計分發層，並分別訂定各訂閱者的緩衝策略。最後保留測試程式碼、工具鏈版本與業務契約說明，讓之後接入事件來源的人知道，哪些丟棄是設計的一部分，哪些屬於故障。

常見問題