Concurrent iteration of AsyncStream can 'leak' continuations

[jamieQ]

Description

per its documention, AsyncStream nominally does not support concurrent iteration. in practice, however, this functionality exists and can be used. if one creates multiple iterators over a single underlying AsyncStream and uses them from different Tasks, then termination of the stream may cause some number of the concurrent iterators to remain suspended indefinitely.

Reproduction

func test_multiConsume() async {
    let (stream, continuation) = AsyncStream<Int>.makeStream()

    await withTaskGroup(of: Void.self) { group in
        for i in 1...10 { // fewer consumers will often also result in the same issue
            group.addTask {
                print("starting iterator \(i)")
                for await value in stream {
                    print("iterator \(i): \(value)")
                }
                print("iterator \(i) finished")
            }
        }

        group.addTask {
            print("terminating stream")
            continuation.finish()
        }
    }
    print("done") // generally never reached
}

Expected behavior

AsyncStream should not allow concurrent iteration. attempting to do so should be either a compiler or runtime error, as it (sometimes) is with AsyncThrowingStream.

Environment

swift-driver version: 1.87.3 Apple Swift version 5.9.2 (swiftlang-5.9.2.2.56 clang-1500.1.0.2.5)
Target: arm64-apple-macosx13.0

Additional information

related forums discussion
related existing bug report