AsyncStream in Swift: Stop Buffering Events You Don't Need

The download is already at 80 percent, but your progress bar is still working through updates from 15 percent. Memory keeps climbing. Nothing is obviously blocked, and every callback is being delivered exactly as requested.

That last detail is the problem.

When you bridge a callback API into AsyncStream, you also choose what happens when production outruns consumption. Leaving the default buffer in place can turn a stream of useful information into a queue of obsolete work.

For a progress indicator, the newest snapshot usually matters. For a file’s byte chunks, every value matters. Those are different contracts, even if both happen to use for await.

Let’s build a progress adapter that makes that distinction explicit, handles termination safely, and can be tested without guessing how long a task needs to run.

1. Decide Whether Each Event Is a Snapshot or a Command

Start with the meaning of the value, not the stream initializer.

struct DownloadProgress: Sendable, Equatable {
    let bytesReceived: Int64
    let expectedBytes: Int64?

    var fraction: Double? {
        guard let total = expectedBytes, total > 0 else { return nil }
        return min(1, max(0, Double(bytesReceived) / Double(total)))
    }
}

This is a complete snapshot. If you receive 100 KB, then 200 KB, displaying only the second value is still correct. An unknown total remains unknown rather than becoming a misleading zero-percent reading.

Now imagine the callback instead supplies Data containing the next chunk. Dropping the first chunk corrupts the file. A counter increment, payment instruction, or database mutation has the same problem: the next event does not replace the previous one.

Write this distinction into your adapter’s documentation. It is easy for a later refactor to replace a snapshot with a delta while leaving the buffering policy untouched. The code will compile, and the behavior will quietly become wrong.

For search, the distinction gets another layer. Keeping the newest result snapshot does not prove that it belongs to the newest query. You still need query identity or cancellation of superseded requests. Buffering cannot repair an upstream ordering bug.

2. Choose a Policy You Can Explain

The three policies make different promises about pending elements. Apple’s buffering-policy documentation is the API reference for these choices.

For our progress UI, choose .bufferingNewest(1). That is one pending snapshot, not one total value in the system. The consumer can already be processing another value.

A newest-value buffer also isn’t a display-rate limiter. If the consumer is ready for every callback, it can receive every callback. If you want updates at a particular cadence, apply a separate sampling or throttling design appropriate to your UI.

Be careful with .bufferingOldest(1): when the consumer is slow, it preserves the stale pending reading instead of replacing it with a current one. That may be exactly right for a first-occurrence diagnostic, but it is rarely what a progress indicator wants.

Capacity zero is another distinct choice. Without an awaiting consumer, the value is dropped. It does not create a rendezvous where the producer waits.

The API came with Swift 5.5 through SE-0314. The examples here were checked with Swift 6.4 in Swift 6 language mode. The adapter uses Synchronization.Mutex, so its Apple-platform baseline is iOS 18 or macOS 15; older deployment targets need an equivalent audited synchronization primitive.

3. Inspect Dropped Values Without Creating More Work

yield returns synchronously. Its result tells you what happened to that emission; it does not acknowledge that your UI finished displaying anything. See the continuation’s yield documentation.

func recordProgress(
    _ value: DownloadProgress,
    into continuation: AsyncStream<DownloadProgress>.Continuation
) {
    switch continuation.yield(value) {
    case .enqueued:
        break
    case .dropped(let discarded):
        // Example only: replace with inexpensive aggregated diagnostics.
        print("Discarded snapshot at \(discarded.bytesReceived) bytes")
    case .terminated:
        break
    @unknown default:
        break
    }
}

With a newest-value buffer, discarded can be the older buffered snapshot, not the value you just submitted. With an oldest-value buffer, it is the rejected new arrival. That distinction matters when you attach identifiers to diagnostic events.

Avoid logging every drop in production. A high-frequency source plus synchronous logging can make the callback path more expensive than the UI it feeds. Count or sample losses instead, and record enough context to distinguish intentional coalescing from unexpected overload.

Also avoid interpreting .enqueued(remaining:) as permission for future production. Another callback may change the buffer immediately. It is a result of this operation, not a reservation.

4. Make the Subscription’s Lifetime Explicit

Our callback source is an application-defined boundary, not an invented URLSession API. A real downloader or delegate wrapper implements it.

import Synchronization

protocol DownloadProgressSource: Sendable {
    // Deliver ordered snapshots, then exactly one completion.
    // The returned closure must be thread-safe, idempotent, and
    // release the registered callbacks when cancellation completes.
    func observeProgress(
        _ progress: @escaping @Sendable (DownloadProgress) -> Void,
        completion: @escaping @Sendable (Result<Void, any Error>) -> Void
    ) -> @Sendable () -> Void
}

final class CancellationGate: Sendable {
    private struct State {
        var stopped = false
        var action: (@Sendable () -> Void)?
    }

    private let state = Mutex(State())

    // Called exactly once, after registration returns its token.
    func install(_ action: @escaping @Sendable () -> Void) {
        let stopImmediately = state.withLock { state in
            if state.stopped { return true }
            state.action = action
            return false
        }
        if stopImmediately { action() }
    }

    func cancel() {
        let action = state.withLock { state in
            state.stopped = true
            let action = state.action
            state.action = nil
            return action
        }
        action?()
    }
}

struct ProgressObservation: Sendable {
    let values: AsyncThrowingStream<DownloadProgress, any Error>
    let cancel: @Sendable () -> Void
}

func observeDownload(
    from source: any DownloadProgressSource
) -> ProgressObservation {
    let gate = CancellationGate()
    let pair = AsyncThrowingStream<DownloadProgress, any Error>.makeStream(
        bufferingPolicy: .bufferingNewest(1)
    )

    pair.continuation.onTermination = { _ in gate.cancel() }

    let stop = source.observeProgress(
        { snapshot in
            pair.continuation.yield(snapshot)
        },
        completion: { result in
            switch result {
            case .success:
                pair.continuation.finish()
            case .failure(let error):
                pair.continuation.finish(throwing: error)
            }
        }
    )
    gate.install(stop)

    return ProgressObservation(values: pair.stream, cancel: {
        gate.cancel()
        pair.continuation.finish()
    })
}

The slightly unglamorous gate handles a real registration race. A source may complete synchronously inside observeProgress, before its cancellation closure has been returned. Termination then marks the gate stopped. When registration finally returns, install immediately invokes the closure instead of storing an already-obsolete subscription.

If registration wins first, cancel removes the closure under the lock and invokes it afterward. Repeated cancellation calls find no action to invoke. Most importantly, no external code executes while the mutex is held. A source’s cancellation callback may reenter the adapter; calling it under the lock risks deadlock.

The source contract is important here. If its cancellation closure requires a particular queue, the source must provide a thread-safe wrapper that dispatches to that queue. The adapter cannot safely assume termination runs on the main actor. Apple’s onTermination documentation describes this synchronous, sendable cleanup hook.

This is also where ownership deserves attention. The subscription can retain its callbacks, and those callbacks retain the continuation. Cancellation must tear down that registration. If you are chasing unexpectedly retained screens, review the same ownership paths discussed in iOS ARC and retain-cycle debugging.

The stream starts observing when observeDownload runs, not when iteration begins. Build it close to its consumption scope. If you store it for later, you own the responsibility to cancel it even when nobody ever iterates.

5. Consume on the Right Actor, and Cancel on Exit

The UI should own the observation rather than assume that leaving a loop cleans up an external producer.

@MainActor
final class DownloadProgressModel {
    private(set) var fraction: Double?
    private(set) var failureMessage: String?

    func track(_ source: any DownloadProgressSource) async {
        let observation = observeDownload(from: source)
        defer { observation.cancel() }

        do {
            for try await snapshot in observation.values {
                try Task.checkCancellation()
                fraction = snapshot.fraction
            }
        } catch is CancellationError {
            // The owning screen's task stopped consuming.
        } catch {
            if !Task.isCancelled {
                failureMessage = String(describing: error)
            }
        }
    }
}

Call track from a lifecycle-owned task. If you wrap this model with Observation for SwiftUI, expose the properties using the state-management approach appropriate to your app.

defer covers normal completion, errors, and your own early exit. onTermination covers the stream’s termination path, including cancellation during iteration. They cooperate through the idempotent gate.

Do not rely on break alone to synchronously unregister a callback. A retained continuation or stored stream can outlive that loop. Likewise, cancellation during some unrelated long-running operation in the loop body does not magically cancel that operation or immediately run defer. Keep consumer work cancellation-cooperative. The async cleanup and cancellation guide covers the broader resource-lifetime problem.

Progress is not the download’s result. A 100-percent snapshot does not prove the file was moved successfully or validated. Keep the authoritative downloaded-file result in your download API. Here, normal stream completion means observation ended normally; cancellation can also end iteration, so never treat an empty loop exit as proof of success.

6. Bounded Buffering Is Not Backpressure

We have bounded the stream’s pending snapshot count. We have not slowed the downloader. It still invokes every callback, and the adapter still executes every yield.

If creating each snapshot requires expensive decoding, copying, or allocation, a one-slot buffer does not remove that upstream cost. Move coalescing earlier when possible. Avoid dispatching one unstructured task per callback: you may simply replace an element queue with an uncontrolled task queue.

When a producer can genuinely suspend, a pull-based sequence or Swift Async Algorithms’ AsyncChannel may be more appropriate. A channel’s send suspends according to consumer demand; it does not wait for all processing inside the consumer’s loop body to finish. See the official channel guide.

Putting await channel.send inside a newly created task for every synchronous callback defeats that purpose: the callback source still produces freely, leaving tasks waiting to send. You need cooperation from the actual producer, such as pausing reads or requesting the next item only when ready.

For lossless workloads with a non-pausable producer, decide explicitly what overload means: persist to disk, reject work, disconnect, or fail with a capacity error. Unlimited buffering is a capacity decision, not a reliability guarantee.

Keep one consuming task per observation. Treat this adapter as single-consumer; it is not a broadcast subscription for several screens. Create separate subscriptions or design a dedicated fan-out layer when that is what the product needs.

7. Test the Contract, Not the Scheduler

You do not need to sleep to create a full buffer. Submit values before making the first iterator request.

import Testing
import Synchronization

@Test func newestBufferReplacesThePendingValue() async {
    let pair = AsyncStream<Int>.makeStream(bufferingPolicy: .bufferingNewest(1))
    pair.continuation.yield(10)
    let result = pair.continuation.yield(20)
    pair.continuation.finish()

    if case .dropped(let value) = result {
        #expect(value == 10)
    } else {
        Issue.record("Expected replacement of the older snapshot")
    }

    var iterator = pair.stream.makeAsyncIterator()
    let first = await iterator.next()
    let end = await iterator.next()
    #expect(first == 20)
    #expect(end == nil)
}

@Test func oldestBufferRejectsTheNewArrival() async {
    let pair = AsyncStream<Int>.makeStream(bufferingPolicy: .bufferingOldest(1))
    pair.continuation.yield(10)
    let result = pair.continuation.yield(20)
    pair.continuation.finish()

    if case .dropped(let value) = result {
        #expect(value == 20)
    } else {
        Issue.record("Expected rejection of the newer snapshot")
    }

    var iterator = pair.stream.makeAsyncIterator()
    let first = await iterator.next()
    let end = await iterator.next()
    #expect(first == 10)
    #expect(end == nil)
}

@Test func terminationBeforeRegistrationStillCancelsOnce() {
    let calls = Mutex(0)
    let gate = CancellationGate()
    gate.cancel()
    gate.install { calls.withLock { $0 += 1 } }
    gate.cancel()
    #expect(calls.withLock { $0 } == 1)
}

@Test func cancellingAConsumerStopsItsProducer() async {
    let calls = Mutex(0)
    let gate = CancellationGate()
    gate.install { calls.withLock { $0 += 1 } }
    let pair = AsyncStream<Int>.makeStream()
    pair.continuation.onTermination = { _ in gate.cancel() }

    let consumer = Task {
        var iterator = pair.stream.makeAsyncIterator()
        return await iterator.next()
    }
    consumer.cancel()
    let value = await consumer.value
    #expect(value == nil)
    #expect(calls.withLock { $0 } == 1)
}

The cancellation test is valid whether cancellation wins before next() or while it is suspended. There are no buffered elements, so neither scheduling order permits a value. Awaiting task completion establishes when the termination check can be made.

These tests target distinct invariants: replacement, rejection, the late-registration path, and consumer cancellation. Add integration tests against your real source for completion errors, callback release, and callbacks racing with explicit cancellation. A source requiring queue-specific cleanup needs tests for that contract too.

For more on organizing these as small, focused tests, see the practical Swift Testing guide.

8. Conclusion

An async sequence is an interface. It does not decide whether data is replaceable, stop an external subscription for you, or make a fast producer respect a slow consumer.

For download progress, a complete snapshot and a newest-value buffer are usually a good fit. Pair them with an explicit owner, race-safe cancellation, and tests that exercise the contract directly. For lossless data, change the transport or overload strategy instead of hoping a different buffer size will make the problem disappear.

Key Takeaways

• Drop stale snapshots only when newer values completely replace them.
• .bufferingNewest(1) bounds pending values; it does not throttle callbacks or apply backpressure.
• Inspect the actual dropped value, and keep overload diagnostics inexpensive.
• Install termination handling before registering callbacks, and handle a token returned after termination.
• Cancel explicitly when consumption ends, and make the source release its callbacks.
• Test controlled production and lifetime transitions without arbitrary delays.