Test AsyncStream buffering and termination on a cloud Mac

When you feed download progress, device status, or notification callbacks into AsyncStream, a UI that updates correctly does not prove the event stream is behaving correctly. A slightly slower consumer can cause buffered values to be dropped, and the underlying subscription may keep running after the user leaves the page. On a VpsKeep cloud Mac, you can establish these behaviors with standalone Swift tests before wiring the stream back into your app, rather than hunting for races by repeatedly clicking through the UI.

Decide which events may be dropped

Progress percentages usually need only the latest value, making bufferingNewest a good fit. If you need to retain the events that arrived first, choose bufferingOldest. Neither policy provides a lossless queue. Data that must not be lost, such as order changes or file-write confirmations, needs an additional acknowledgment or persistence mechanism.

The default unbounded buffer is not a universal solution either. If production consistently outpaces consumption, queued elements will keep consuming memory. Define three constraints first: buffer capacity, which elements to retain when it is full, and when to release the subscription. Set capacity according to the backlog your application can tolerate, not by arbitrarily increasing it to match the machine's available memory.

AsyncStream's yield does not wait for the consumer to make room. A bounded buffer provides a dropping policy, not automatic backpressure.

This article verifies the rules with an integer stream whose capacity is 2. The integers represent event sequence numbers, not performance measurements.

Create a standalone test workspace

Run the following commands in a writable directory to create a test package using the currently selected development toolchain. Record the Swift version and use the same toolchain for subsequent runs. The package needs neither application secrets nor connections to real services.

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

Save the code in the next section as Tests/StreamContractTests/StreamContractTests.swift. If initialization generated a file with that name, replace it. You can keep any other tests generated by the template; the command below filters for this test class only.

To make the overflow result deterministic, the tests synchronously produce three elements before starting consumption. Do not let the producer and consumer compete for scheduling, and do not use a one-second wait to guess whether the buffer is full. These tests verify the container's contract, not thread execution speed.

Check both dropped values and the retained sequence

Asserting only the final array is not enough. The code should also check the return value of the third yield: when retaining the newest elements, the old element 1 is dropped; when retaining the oldest elements, the new element 3 is dropped. After finish is called, values already in the buffer can still be read.

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

When production code handles .dropped, it should count dropped values or trigger compensating application logic, rather than synchronously writing a log entry for every progress event. When handling .terminated, stop delivering values instead of repeatedly retrying the same continuation.

Verify that cancellation triggers the unsubscribe hook

Append the following extension to the same file. The stream contains no elements, so the consumer can only wait or respond to cancellation. A test expectation verifies the termination notification without relying on a fixed sleep. The two-second timeout is only a deadline for failure, not a delay between execution steps.

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

In a real adapter, replace the callback that fulfills the expectation with the method that releases the subscription token. Termination handling is not guaranteed to run on the main thread, so it must not update the UI directly. Unsubscribing must be thread-safe and idempotent. Also, do not call an unsubscribe function that could trigger a callback while holding the adapter's internal lock.

Account for a race during subscription setup as well: termination may happen before the token is returned. The adapter needs to retain a “closed” state. If a token arrives after that state has been set, release it immediately. Setting onTermination alone does not cover this path.

Add the tests to your pipeline and review delivery checks

Run the tests and save the output in the pipeline's standard test records. With fixed inputs, repeated runs should produce the same arrays and dropped elements. Intermittent passes usually indicate that real timers or shared event sources have slipped into the tests.

swift test --filter StreamContractTests

Before wiring the stream back into the app, complete four checks: whether a normal finish unsubscribes; whether task cancellation unsubscribes; whether late callbacks during unsubscription are ignored; and whether a token arriving after termination is released immediately. Test these paths with a fake event source that lets you deliver events manually, then test the real source.

Do not simulate broadcasting by having multiple tasks compete to read the same stream. If you need multiple subscribers, explicitly design a distribution layer and define each subscriber's buffering policy separately. Retain the test code, toolchain version, and application contract documentation so that anyone connecting an event source later knows which dropped values are intentional and which indicate a failure.

Frequently asked questions