云端 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 是否退订；任务取消是否退订；退订过程中迟到的回调是否被忽略；令牌晚于终止到达时是否立即释放。先用可手动投递的假事件源测试这些路径，再测试真实源。

不要让多个任务竞争读取同一个流来模拟广播。需要多个订阅者时，应明确设计分发层，并分别规定各订阅者的缓冲策略。最终保留测试代码、工具链版本和业务契约说明，让后来接入事件源的人知道哪些丢弃属于设计，哪些属于故障。

常见问题